OpenF1 API - Real-time and historical Formula 1 data
"""The description of each attribute in the `API methods` section of the documentation."""


ATTRIBUTES_DESCRIPTIONS = {
    'date_start': 'The starting date and time, in ISO 8601 format.',
    'date_end': 'The ending date and time, in ISO 8601 format.',
    'driver_number': 'The unique number assigned to an F1 driver (cf. <a href="https://en.wikipedia.org/wiki/List_of_Formula_One_driver_numbers#Formula_One_driver_numbers" target="_blank">Wikipedia</a>).',
    'duration_sector_1': 'The time taken, in seconds, to complete the first sector of the lap.',
    'duration_sector_2': 'The time taken, in seconds, to complete the second sector of the lap.',
    'duration_sector_3': 'The time taken, in seconds, to complete the third sector of the lap.',
    'i1_speed': 'The speed of the car, in km/h, at the first intermediate point on the track.',
    'i2_speed': 'The speed of the car, in km/h, at the second intermediate point on the track.',
    'is_pit_out_lap': 'A boolean value indicating whether the lap is an "out lap" from the pit (`true` if it is, `false` otherwise).',
    'lap_duration': 'The total time taken, in seconds, to complete the entire lap.',
    'lap_number': 'The sequential number of the lap within the session (starts at 1).',
    'meeting_key': 'The unique identifier for the meeting. Use `latest` to identify the latest or current meeting.',
    'segments_sector_1': 'A list of values representing the "mini-sectors" within the first sector (see mapping table below).',
    'segments_sector_2': 'A list of values representing the "mini-sectors" within the second sector (see mapping table below).',
    'segments_sector_3': 'A list of values representing the "mini-sectors" within the third sector (see mapping table below).',
    'session_key': 'The unique identifier for the session. Use `latest` to identify the latest or current session.',
    'st_speed': 'The speed of the car, in km/h, at the speed trap, which is a specific point on the track where the highest speeds are usually recorded.',
    'date': 'The date and time, in ISO 8601 format.',
    'x': "The 'x' value in a 3D Cartesian coordinate system representing the current approximate location of the car on the track.",
    'y': "The 'y' value in a 3D Cartesian coordinate system representing the current approximate location of the car on the track.",
    'z': "The 'z' value in a 3D Cartesian coordinate system representing the current approximate location of the car on the track.",
    'circuit_key': 'The unique identifier for the circuit where the event takes place.',
    'circuit_short_name': 'The short or common name of the circuit where the event takes place.',
    'country_code': 'A code that uniquely identifies the country.',
    'country_key': 'The unique identifier for the country where the event takes place.',
    'country_name': 'The full name of the country where the event takes place.',
    'gmt_offset': 'The difference in hours and minutes between local time at the location of the event and Greenwich Mean Time (GMT).',
    'location': 'The city or geographical location where the event takes place.',
    'session_name': 'The name of the session (`Practice 1`, `Qualifying`, `Race`, ...).',
    'session_type': 'The type of the session (`Practice`, `Qualifying`, `Race`, ...).',
    'year': 'The year the event takes place.',
    'compound': 'The specific compound of tyre used during the stint (`SOFT`, `MEDIUM`, `HARD`, ...).',
    'stint_number': 'The sequential number of the stint within the session (starts at 1).',
    'lap_start': 'Number of the initial lap in this stint (starts at 1).',
    'lap_end': 'Number of the last completed lap in this stint.',
    'tyre_age_at_start': 'The age of the tyres at the start of the stint, in laps completed.',
    'recording_url': 'URL of the radio recording.',
    'meeting_name': 'The name of the meeting.',
    "meeting_official_name": 'The official name of the meeting.',
    'broadcast_name': "The driver's name, as displayed on TV.",
    'first_name': "The driver's first name.",
    'full_name': "The driver's full name.",
    'headshot_url': "URL of the driver's face photo.",
    'last_name': "The driver's last name.",
    'team_colour': "The hexadecimal color value (RRGGBB) of the driver's team.",
    'team_name': "Name of the driver's team.",
    'name_acronym': "Three-letter acronym of the driver's name.",
    'rpm': 'Revolutions per minute of the engine.',
    'speed': 'Velocity of the car in km/h.',
    'n_gear': "Current gear selection, ranging from 1 to 8. `0` indicates neutral or no gear engaged.",
    'throttle': 'Percentage of maximum engine power being used.',
    'drs': 'The Drag Reduction System (DRS) status (see mapping table below).',
    'brake': "Whether the brake pedal is pressed (`100`) or not (`0`).",
    'air_temperature': 'Air temperature (°C).',
    'humidity': 'Relative humidity (%).',
    'pressure': 'Air pressure (mbar).',
    'rainfall': 'Whether there is rainfall.',
    'track_temperature': 'Track temperature (°C).',
    'wind_direction': 'Wind direction (°), from 0° to 359°.',
    'wind_speed': 'Wind speed (m/s).',
    'category': 'The category of the event (`CarEvent`, `Drs`, `Flag`, `SafetyCar`, ...).', 
    'flag': 'Type of flag displayed (`GREEN`, `YELLOW`, `DOUBLE YELLOW`, `CHEQUERED`, ...).',
    'message': 'Description of the event or action.',
    'scope': 'The scope of the event (`Track`, `Driver`, `Sector`, ...).',
    'sector': 'Segment ("mini-sector") of the track where the event occurred? (starts at 1).',
    'position': 'Position of the driver (starts at 1).',
    'gap_to_leader': 'The time gap to the race leader in seconds, `+1 LAP` if lapped, or `null` for the race leader.',
    'interval': 'The time gap to the car ahead in seconds, `+1 LAP` if lapped, or `null` for the race leader.',
    'pit_duration': 'The time spent in the pit, from entering to leaving the pit lane, in seconds.',
}

documentation/attributes_descriptions.py
br-g/openf1
"""A script to dynamically discover API methods and generate the `API methods` section of the documentation.    
   Automatically executed by Github Actions after each commit on branch `main`.
"""

import sys
import html
import requests
import json
from pathlib import Path
from attributes_descriptions import ATTRIBUTES_DESCRIPTIONS

sys.path.append('../query_api')
from util import get_api_methods

METHODS_DIR = Path('../query_api/methods')
OUTPUT_FILE = Path('source/includes/_api_methods.md')


def _get_method_documentation(url: str, method: 'BaseMethod') -> str:
    """Generates documentation for a given API method"""
    name = url.split('/')[-1]
    url = f'https://api.openf1.org/{url}'
    query_params = '&'.join(method.example_filters) if method.example_filters else ''

    attributes = (
        'Name | Description\n'
        + '--------- | -----------\n'
        + '\n'.join([f'{name} | {ATTRIBUTES_DESCRIPTIONS.get(name, "")}' for name in sorted(list(method.attributes))])
    )

    output = json.dumps(method.example_response, indent=2, sort_keys=True)
    url_full = f'{url}{"?" if query_params else ""}{query_params}'

    return f'''
## {name.capitalize().replace('_', ' ')}

{method.description}

```shell
curl "{url_full}"
```

```python
from urllib.request import urlopen
import json

response = urlopen('{url_full}')
data = json.loads(response.read().decode('utf-8'))
print(data)

# If you want, you can import the results in a DataFrame (you need to install the `pandas` package first)
# import pandas as pd
# df = pd.DataFrame(data)
```

```r
# If needed, install libraries
# install.packages('httr')
# install.packages('jsonlite')

library(httr)
library(jsonlite)

response <- GET('{url_full}')
parsed_data <- fromJSON(content(response, 'text'))
print(parsed_data)

# If you want, you can import the results in a DataFrame
# df <- do.call(rbind, lapply(parsed_data, data.frame, stringsAsFactors = FALSE))
# df <- as.data.frame(t(as.matrix(df)))
```

```javascript
fetch('{url_full}')
  .then(response => response.json())
  .then(jsonContent => console.log(jsonContent));
```

> Output:

```json
{output}
```

### HTTP Request
`GET {html.escape(url)}`

### Sample URL
<a href="{html.escape(url_full)}" target="_blank">{html.escape(url_full)}</a>

### Attributes
{attributes}

{"<br /><br />" if method.additional_info else ""}
{method.additional_info}
'''


def generate_documentation() -> None:
    """Generates the `API methods` section of the documentation and writes it to
       the source folder."""
    markdown = ['# API methods']
    for url, cls in sorted(list(get_api_methods(METHODS_DIR))):
        markdown.append(
            _get_method_documentation(url=url, method=cls)
        )

    with open(OUTPUT_FILE, 'w') as f:
        f.write('\n'.join(markdown))


if __name__ == '__main__':
    generate_documentation()

documentation/generate.py
br-g/openf1
import os
from datetime import datetime, timedelta
import tempfile
import threading
import traceback
import pytz
import asyncio
from flask import Flask
from util import upload_blob, get_shedule
from logger import logger
from ingest_real_time import tail_and_ingest

FILEKEY_RAW = '%Y/%m/%d/%H:%M:%S.txt'
RAW_UPLOAD_FREQUENCY = 60  # in seconds
RECORDING_DURATION = 13 * 60  # in seconds


async def upload_raw_recordings(filepath: str, filekey_raw: str):
    """Uploads raw data to a GCS bucket, at regular intervals, as a backup"""
    while True:
        await asyncio.sleep(RAW_UPLOAD_FREQUENCY)

        loop = asyncio.get_running_loop()
        await loop.run_in_executor(
            None,
            upload_blob,
            os.environ.get('BUCKET_API_RAW'),
            filepath,
            filekey_raw,
        )


def is_session_possibly_in_progress(offset_before=timedelta(hours=2), offset_after=timedelta(hours=5)) -> bool:
    """Returns whether it is worth listening to data coming from the F1 server.
       Recording starts long before the session starts and ends long after, just in case...
    """
    time_ = datetime.now().astimezone(pytz.timezone('UTC'))

    for session in get_shedule(time_.year):
        if session['SessionDateUtc'] - offset_before <= time_ <= session['SessionDateUtc'] + offset_after:
            return True
    
    return False


async def record(filepath: str):
    """Records raw F1 data to a file, using the FastF1 live timing module
       (https://github.com/theOehrly/Fast-F1/tree/master/fastf1/livetiming)
    """
    start_time = datetime.now(pytz.utc)

    command = ['python', '-m', 'fastf1.livetiming', 'save', filepath, '--timeout', str(RECORDING_DURATION)]
    logger.info(f'Starting recording live data (command: `{" ".join(command)}`)')
    process_fastf1 = await asyncio.create_subprocess_exec(
        *command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    # Wait for the planned recording duration
    while datetime.now(pytz.utc) < start_time + timedelta(seconds=RECORDING_DURATION):
        await asyncio.sleep(5)
    
    logger.info(f'Timing out after {round(RECORDING_DURATION/60,2)} minutes.')

    # Stop recording
    process_fastf1.terminate()
    try:
        await asyncio.wait_for(process_fastf1.wait(), timeout=10)
    except asyncio.TimeoutError:
        logger.warning('Process did not terminate in time. Killing.')
        process_fastf1.kill()
    
    logger.info('Recording stopped.')


def _handle_exception(task: asyncio.Task):
    """Logs exceptions that occur in asynchronous jobs"""
    exception = task.exception()
    if exception is not None:
        # Skip exception that is raised at the end of the task
        if isinstance(exception, asyncio.exceptions.CancelledError):
            return
        tb_str = ''.join(traceback.format_exception(etype=type(exception), value=exception, tb=exception.__traceback__))
        logger.error(tb_str)


async def run():
    """Records session if one could be in progress"""
    if not is_session_possibly_in_progress():
        logger.info('No session in progress according to schedule. Skipping.')
        return

    filekey_raw = datetime.utcnow().strftime(FILEKEY_RAW)
    
    logger.info('A session could be in progress according to schedule. Recording.')
    with tempfile.NamedTemporaryFile(mode='w', delete=True) as temp:
        task1 = asyncio.create_task(upload_raw_recordings(temp.name, filekey_raw))
        task1.add_done_callback(_handle_exception)
        task2 = asyncio.create_task(tail_and_ingest(temp.name))
        task2.add_done_callback(_handle_exception)
        await record(temp.name)


def _start_async_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(run())


app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def main():
    logger.info('Ingestor started')
    t = threading.Thread(target=_start_async_loop)
    t.start()
    t.join()
    return 'Completed', 200


if __name__ == '__main__':
    app.run(debug=False, host='0.0.0.0', port=8080)

ingestor/app.py
br-g/openf1
from typing import List, Dict, Iterator, Optional
import os
import json
import hashlib
import asyncio
from datetime import datetime
import pymongo
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
from logger import logger

BATCH_SIZE = 1000  # Maximum number of JSON docs to be uploaded at once to MongoDB


def get_mongo_db() -> AsyncIOMotorClient:
    """Returns an asynchronous MongoDB client.
       The client could not be a global variable because it is not thread safe.
    """
    return AsyncIOMotorClient(os.getenv('MONGO_CONNECTION_STRING'))['raw']


collection_names = set()
async def _collection_exists(name: str, mongo_db: Optional[AsyncIOMotorClient] = None) -> bool:
    """Returns whether a collection exists in the MongoDB database"""
    if mongo_db is None:
        mongo_db = get_mongo_db()

    global collection_names
    if name in collection_names:
        return True
    else:
        collection_names = set(await mongo_db.list_collection_names())
        return name in collection_names


def _time_to_str(dct: Dict) -> Dict:
    """Turns datetime objects into ISO strings"""
    res = {}
    for key, value in dct.items():
        if isinstance(value, dict):
            res[key] = _time_to_str(value)
        elif isinstance(value, datetime):
            res[key] = value.isoformat()
        else:
            res[key] = value
    return res


def _hash_dict(dct: Dict) -> str:
    """Hashes dict content"""
    dct_str = json.dumps(_time_to_str(dct), sort_keys=True)
    return hashlib.sha256(dct_str.encode()).hexdigest()


async def _insert(collection: AsyncIOMotorCollection, docs: List[Dict]):
    """Inserts documents to a MongoDB collection and skips duplicates"""
    # Create an ID to check for duplicates
    for doc in docs:
        doc['_id'] = doc['_time'].strftime('%y%m%d%H%M%S') + _hash_dict(doc)[:4]  # limit ID length to reduce storage footprint

    try:
        await collection.insert_many(docs, ordered=False)
    except pymongo.errors.BulkWriteError as e:
        # Handle the duplicate key errors (and possibly other errors)
        # Duplicates are expected as there are multiple ingestor instances working
        # simultaneously (to increase resilience).
        for error in e.details.get('writeErrors'):
            if error.get('code') == 11000:  # Duplicate key error code
                pass
            else:
                logger.error(error)


def _get_batches(docs: List[Dict]) -> Iterator[List[Dict]]:
    """Yields batches of maximum size `BATCH_SIZE`"""
    for i in range(0, len(docs), BATCH_SIZE):
        yield docs[i:i+BATCH_SIZE]


async def insert_data_async(collection_name: str, docs: List[Dict],
                            mongo_db: Optional[AsyncIOMotorClient] = None):
    """Inserts documents to a MongoDB collection, in batches, and skips duplicates."""
    if mongo_db is None:
        mongo_db = get_mongo_db()

    collection = mongo_db[collection_name]
    tasks = [_insert(collection=collection, docs=batch) for batch in _get_batches(docs)]
    await asyncio.gather(*tasks)

ingestor/db.py
br-g/openf1
"""Script for parsing and saving raw historical F1 data to database."""

from typing import Tuple, Dict, Optional, List, Iterator
import os
import re
import json
from collections import defaultdict
import requests
from datetime import datetime, timedelta
import asyncio
import click
from tqdm import tqdm
from parsing import parse_line
from db import insert_data_async, get_mongo_db
from logger import logger
from util import to_timedelta

BASE_API_URL = 'https://livetiming.formula1.com/static'


def _download_topic(session_path: str, filename: str) -> str:
    """Downloads raw data for a specific topic"""
    url = os.path.join(BASE_API_URL, session_path, filename)
    return requests.get(url).text


def _get_schedule(year: int) -> Dict:
    """Returns data about all the past sessions for a specific year"""
    url = os.path.join(BASE_API_URL, f'{year}/Index.json')
    return json.loads(requests.get(url).content)


def _get_session_keys(year: int, meeting_key: int) -> List[int]:
    """Returns the keys of the sessions for a specific meeting"""
    for meeting in _get_schedule(year)['Meetings']:
        if meeting['Key'] == meeting_key:
            return [e['Key'] for e in meeting['Sessions']]
            
    raise SystemError(f'Meeting not found (year: `{year}`, meeting_key: `{meeting_key}`).')


def _parse(line: str) -> Tuple[timedelta, str]:
    """Parses a line to extract the duration since session start and raw data.

    The line is expected to be formatted as follows:
    (duration since session start, raw data)
    """
    pattern = r'(\d+:\d+:\d+\.\d+)(.*)'
    match = re.match(pattern, line)
    return to_timedelta(match.group(1)), match.group(2).strip('\r').strip('"')


def parse_topic(session_path: str, topic_filename: str, session_key: Optional[int] = None,
                meeting_key: Optional[int] = None, t0: Optional[datetime] = None) -> Dict[str, List[Dict]]:
    """Parses a topic file for a specified session and returns the parsed data as a dictionary.
       `t0` is the start time of the session when data begins to stream.
    """
    res = defaultdict(list)

    raw = _download_topic(session_path=session_path, filename=topic_filename)
    if '<Error><Code>NoSuchKey</Code>' in raw:
        raise SystemError(f'Topic not found at `{session_path}`.')

    for line in raw.split('\n'):
        if not line:
            continue
        session_time, content = _parse(line)
        parsed = parse_line(
            topic=os.path.splitext(topic_filename)[0],
            content=content,
            session_key=session_key,
            meeting_key=meeting_key,
            time=t0 + session_time if t0 else None,
            session_time=session_time.total_seconds(),
        )
        # Group data by collection for faster addition to the database
        for collection, vals in parsed.items():
            res[collection] += vals

    return dict(res)


def get_session_path(year: int, meeting_key: int, session_key: int) -> str:
    """Retrieves the session path from the F1 "Index.json" data"""
    for meeting in _get_schedule(year)['Meetings']:
        if meeting['Key'] == meeting_key:
            for session in meeting['Sessions']:
                if session['Key'] == session_key:
                    return session['Path']
            
    raise SystemError(f'Session not found (year: `{year}`, meeting_key: `{meeting_key}`, ' \
                      f'session_key: `{session_key}`).')


def estimate_t0(session_path: str) -> datetime:
    """Estimates `t0`, the start time of the session when data begins to stream.
       The calculation method comes from the FastF1 package (https://github.com/theOehrly/Fast-F1/blob/317bacf8c61038d7e8d0f48165330167702b349f/fastf1/core.py#L2208).
    """
    position_z = parse_topic(
        session_path=session_path,
        topic_filename='Position.z.jsonStream',
    )
    t0_position_z = max([e['_val']['Timestamp'] - timedelta(seconds=e['_session_time'])
                         for e in position_z['Position.z-Position']])

    car_data = parse_topic(
        session_path=session_path,
        topic_filename='CarData.z.jsonStream',
    )
    t0_car_data = max([e['_val']['Utc'] - timedelta(seconds=e['_session_time'])
                       for e in car_data['CarData.z-Entries']])
    
    return max(t0_position_z, t0_car_data)


def _get_topic_api_filenames(session_path: str) -> Iterator[str]:
    """Yields all the available raw data files for the session"""
    url = os.path.join(BASE_API_URL, session_path, 'Index.json')
    content = json.loads(requests.get(url).content)
    for vals in content['Feeds'].values():
        yield vals['StreamPath']


def parse_all_topics(session_path: str, *args, **kwargs) -> Dict[str, List[Dict]]:
    """Parses all the topics for a specified session and returns the parsed data as a dictionary."""
    topic_api_filenames = list(_get_topic_api_filenames(session_path))

    res = defaultdict(list)
    for api_filename in tqdm(topic_api_filenames):
        parsed = parse_topic(
            session_path=session_path,
            topic_filename=api_filename,
            *args, **kwargs
        )
        # Group data by collection for faster addition to the database
        for collection, vals in parsed.items():
            res[collection] += vals
    
    return dict(res)


@click.group()
def cli():
    pass


async def _ingest_session(year: int, meeting_key: int, session_key: int):
    logger.info('Retrieving session path')
    session_path = get_session_path(year=year, meeting_key=meeting_key, session_key=session_key)

    logger.info('Estimating t0')
    t0 = estimate_t0(session_path)

    logger.info('Parsing topics')
    content = parse_all_topics(
        session_path=session_path,
        session_key=session_key,
        meeting_key=meeting_key,
        t0=t0,
    )

    logger.info('Writing to DB')
    mongo_db = get_mongo_db()
    for collection_name, docs in tqdm(list(content.items())):
        await insert_data_async(collection_name=collection_name, docs=docs, mongo_db=mongo_db)


@cli.command()
@click.argument('year', type=int)
@click.argument('meeting_key', type=int)
@click.argument('session_key', type=int)
def ingest_session(*args, **kwargs):
    """Parses and saves raw F1 data to database for a specific session.
       Example usage: `python ingest_history.py ingest-session 2023 1141 7953`

       The session and meeting keys for 2023, for example, could be found here: https://livetiming.formula1.com/static/2023/Index.json
    """
    asyncio.run(_ingest_session(*args, **kwargs))


@cli.command()
@click.argument('year', type=int)
@click.argument('meeting_key', type=int)
def ingest_meeting(year: int, meeting_key: int):
    """Parses and saves raw F1 data to database for a specific meeting.
       Example usage: `python ingest_history.py ingest-meeting 2023 1141`

       The meeting keys for 2023, for example, could be found here: https://livetiming.formula1.com/static/2023/Index.json
    """
    for session_key in _get_session_keys(year=year, meeting_key=meeting_key):
        logger.info(f'----- Ingesting session `{session_key}` -----')
        asyncio.run(_ingest_session(year=year, meeting_key=meeting_key, session_key=session_key))


if __name__ == '__main__':
    cli()

ingestor/ingest_history.py
br-g/openf1
"""Helper functions for parsing and saving raw F1 data to database, in real-time."""

from typing import List
from collections import defaultdict
from datetime import datetime, timedelta
import asyncio
import pytz
from tqdm import tqdm
from parsing import parse_line
from motor.motor_asyncio import AsyncIOMotorClient
from db import get_mongo_db, insert_data_async
from util import to_datetime

_start_time = None  # Date at which data starts being ingested
_ingestion_queue = []  # Data waiting to be ingested. Data is not ingested right away
                       # because we might be waiting for session information first.

# The `session_key` and `meeting_key` can be determined in 2 ways:
# - By receiving a SessionInfo message which contain these information. There is usually
#   such a message at the beginning of sessions.
# - By querying some data recently added to the database that contain these information.
_session_key = None
_meeting_key = None


async def _try_set_session_info(mongo_db: AsyncIOMotorClient):
    """Tries setting `session_key` and `meeting_key` by copying these values
       from documents added recently to the database
    """
    global _session_key
    global _meeting_key

    # Retrieve some documents that have been written recently
    docs = []
    for collection in {'WeatherData', 'Heartbeat', 'Position.z-Position-Entries'}:
        cursor = mongo_db[collection].find({'_time': {
            '$gte': datetime.now().astimezone(pytz.UTC) - timedelta(minutes=30)
        }}).limit(1000)
        docs = await cursor.to_list(length=1000)
        if docs:
            break

    if docs:
        _session_keys = {e['_session_key'] for e in docs if e['_session_key']}
        _meeting_keys = {e['_meeting_key'] for e in docs if e['_meeting_key']}

        if len(_session_keys) == 1:
            _session_key = list(_session_keys)[0]
        if len(_meeting_keys) == 1:
            _meeting_key = list(_meeting_keys)[0]


def _add_line(line: str):
    """Adds a line to the queue to be later parsed and saved"""
    global _session_key
    global _meeting_key
    global _ingestion_queue

    line = line.strip()
    if not line:
        return
    line_data = eval(line)
    _ingestion_queue.append(line_data)
    
    if line_data[0] == 'SessionInfo':
        _session_key = line_data[1]['Key']
        _meeting_key = line_data[1]['Meeting']['Key']


async def _ingest(mongo_db: AsyncIOMotorClient, force: bool = False) -> List:
    """Parses and saves data from the queue, unless we are still waiting for the
       `session_key` and `meeting_key` variables to be set.
       If `force` is True, data is parsed and saved anyway.
    """
    global _start_time
    global _session_key
    global _meeting_key
    global _ingestion_queue

    if not _ingestion_queue:
        return []

    if _start_time is None:
        _start_time = datetime.now()
    
    if not force and _session_key is None and datetime.now()-_start_time < timedelta(minutes=1):
        return []

    # Parse
    parsed = defaultdict(list)
    while _ingestion_queue:
        data = _ingestion_queue.pop(0)
        _parsed = parse_line(
            topic=data[0],
            content=data[1],
            session_key=_session_key,
            meeting_key=_meeting_key,
            time=to_datetime(data[2]),
        )
        # Group data by collection for faster addition to the database
        for collection, vals in _parsed.items():
            parsed[collection] += vals

    # Save to database
    tasks = []
    for collection_name, docs in parsed.items():
        tasks.append(
            asyncio.create_task(insert_data_async(collection_name=collection_name,
                                                  docs=docs, mongo_db=mongo_db))
        )
    return tasks


async def tail_and_ingest(filepath: str):
    """Ingests every new line that is appended to the file"""
    mongo_db = get_mongo_db()
    await _try_set_session_info(mongo_db)

    if _session_key is None or _meeting_key is None:
        await asyncio.sleep(60)
        await _try_set_session_info(mongo_db)

    with open(filepath, 'r') as file:
        # Move to the end of the file
        file.seek(0, 2)
        while True:
            line = file.readline()
            if not line:
                await asyncio.sleep(0.1)  # Sleep a bit before trying again
                continue
            _add_line(line)
            await _ingest(mongo_db=mongo_db)
    
    await _ingest(mongo_db=mongo_db, force=True)


async def read_and_ingest(filepath: str):
    """Ingests the content of the file, line by line"""
    mongo_db = get_mongo_db()
    await _try_set_session_info(mongo_db)

    with open(filepath, 'r') as file:
        tasks = []
        for line in tqdm(list(file)):
            _add_line(line)
            tasks += await _ingest(mongo_db=mongo_db)
            await asyncio.sleep(0.02)  # Don't go to fast!
    
    tasks += await _ingest(mongo_db=mongo_db, force=True)
    if tasks:
        await asyncio.gather(*tasks)

ingestor/ingest_real_time.py
br-g/openf1
import sys
from loguru import logger

# Initialize logging for Cloud Run
logger.remove()
logger.add(sys.stdout, level='INFO')

ingestor/logger.py
br-g/openf1
"""Helper functions for parsing and transforming raw F1 data.

This module provides a collection of utility functions to decode, 
normalize, and prepare raw F1 data (either historical or real-time)
before indexing into MongoDB. This includes JSON decoding, data type casting,
key modifications, array exploding, and more.
"""

from typing import List, Dict, Iterator, Optional, Union, Any
import json
from functools import lru_cache
from collections import defaultdict
import base64
import zlib
from datetime import datetime
from util import to_datetime


def _decode(raw: str) -> Dict:
    """Decodes raw F1 data from either JSON or Base64 encoded compressed JSON"""
    try:
        return json.loads(raw.strip('"'))
    except:
        s = zlib.decompress(base64.b64decode(raw), -zlib.MAX_WBITS)
        return json.loads(s.decode('utf-8-sig'))


def _is_number(s: Any) -> bool:
    """Checks if the given argument can be converted to a float"""
    try:
        float(s)
        return True
    except ValueError:
        return False

@lru_cache(maxsize=None)
def _is_identifier(s: Any) -> bool:
    """Checks if the given argument is an identifier (a team name, a driver number, ...)"""
    TEAM_NAMES = {
        'Red Bull Racing Honda RBPT',
        'Mercedes',
        'Aston Martin Aramco Mercedes',
        'Ferrari',
        'Alpine Renault',
        'McLaren Mercedes',
        'Haas Ferrari',
        'Alfa Romeo Ferrari',
        'Williams Mercedes',
        'AlphaTauri Honda RBPT',
    }
    return (
        _is_number(s)
        or s in TEAM_NAMES
        or (len(s) >= 2 and s[0] == 'p' and s[1].isupper())
    )

def _add_key(obj: Any, key: Any) -> Any:
    """Utility function to store the key information from a flattened dict"""
    if isinstance(obj, dict):
        key_field = '_key'
        while key_field in set(obj.keys()):
            key_field = '_' + key_field
        obj[key_field] = key
        return obj
    elif isinstance(obj, list):
        return [_add_key(e, key=key) for e in obj]
    else:
        return {
            '_key': key,
            '_val': obj,
        }

def _remove_key_identifiers(obj: Any) -> Any:
    """Transforms `obj` such that there are no more identifiers (team name, driver number, ...)
       as dictionary keys. Such fields would be impossible to query in the database otherwise.
    """
    if isinstance(obj, dict):
        if any(_is_identifier(k) for k in obj):
            return [_add_key(_remove_key_identifiers(v), k) for k, v in obj.items()]
        else:
            return {k: _remove_key_identifiers(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [_remove_key_identifiers(e) for e in obj]
    else:
        return obj


def _explode_arrays(collection_id: str, obj: Any,
                    attributes: Optional[Dict] = None) -> Union[Any, List[Any]]:
    """Transforms `obj` to remove arrays, because they could not be easily indexed
       and retrieved in the database.
    """
    if attributes is None:
        attributes = {}
    else:
        attributes = {f'{k}_': v for k, v in attributes.items()}

    if isinstance(obj, list):
        res = []
        for i, e in enumerate(obj):
            attributes['idx'] = i
            _exploded = _explode_arrays(collection_id, e, attributes=attributes)
            if not isinstance(_exploded, list):
                _exploded = [_exploded]
            for j, _ in enumerate(_exploded):
                _exploded[j] = (_exploded[j][0], {'_val': _exploded[j][1]})
                for k, v in attributes.items():
                    _exploded[j][1][f'_{k}'] = v
            res.extend(_exploded)
        return res

    elif isinstance(obj, dict):
        for k, v in obj.items():
            if not isinstance(v, list) and not isinstance(v, dict):
                attributes[k] = v

        exploded = []
        remaining = {}
        for k, v in obj.items():
            _exploded = _explode_arrays(f'{collection_id}-{k}', v, attributes=attributes)
            if isinstance(_exploded, list):
                exploded.extend(_exploded)
            else:
                remaining[k] = _exploded[1]
        
        if remaining and exploded:
            return [(collection_id, remaining)] + exploded
        elif not remaining and exploded:
            return exploded
        else:
            return (collection_id, remaining)

    else:
        return (collection_id, obj)


def _try_parse_date(s: str) -> Union[datetime, str]:
    """Turns `s` into a datetime object if possible"""
    time = to_datetime(s)
    return time if time is not None else s

def _try_parse_number(s: str) -> Union[int, float, str]:
    """Turns `s` into a number (int or float) if possible"""
    try:
        return int(s)
    except ValueError:
        try:
            return float(s)
        except ValueError:
            return s

def _try_parse_boolean(s: str):
    """Turns `s` into a boolean if possible"""
    if s.lower() == 'true':
        return True
    elif s.lower() == 'false':
        return False
    else:
        return s

@lru_cache(maxsize=None)
def _cast(s: str) -> Union[str, datetime, bool, int, float]:
    """Casts `s` to the most specific type possible (str, datetime, bool, int or float)"""
    s = _try_parse_date(s)
    if isinstance(s, datetime):
        return s
    s = _try_parse_boolean(s)
    if isinstance(s, bool):
        return s
    return _try_parse_number(s)

def cast(obj: Any) -> Any:
    """Casts `obj` to the most specific type possible, recursively"""
    if isinstance(obj, str):
        return _cast(obj)
    elif isinstance(obj, dict):
        return {k: cast(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [cast(e) for e in obj]
    else:
        return obj


def _flatten_list(obj: Any) -> Iterator[Any]:
    if isinstance(obj, list):
        for e in obj:
            yield from _flatten_list(e)
    else:
        yield obj


def _apply_custom_transformations(topic: str, data: Dict) -> None:
    """Transforms some raw data for storage optimization"""
    if topic == 'CarData.z':  # This topic has a very large memory footprint in the database
        for elem in data:
            for entry in elem['Entries']:
                new_cars = []
                for old_car in entry['Cars']:
                    new_car = {k: v for k, v in old_car.items() if k != 'Channels'}
                    for channel in old_car['Channels']:
                        new_car[f'ch_{channel["_key"]}'] = channel['_val']
                    new_cars.append(new_car)
                
                entry['Cars'] = new_cars


def parse_line(topic: str, content: str, session_key: str, meeting_key: str, time: datetime,
               session_time: Optional[float] = None) -> Dict[str, Dict]:
    """Parses a single raw line of F1 data.

    This function decodes and transforms a raw F1 data line into a more
    structured and usable format for indexing into MongoDB.
    """
    decoded = _decode(content) if isinstance(content, str) else content
    flattened = list(_flatten_list(_remove_key_identifiers(decoded)))

    _apply_custom_transformations(topic=topic, data=flattened)

    parsed = defaultdict(list)
    for content in flattened:
        # Explode
        exploded = _explode_arrays(topic, content)
        if not isinstance(exploded, list):
            exploded = [exploded]

        for collection_id, data in exploded:
            if not data:
                continue
            data = cast(data)

            # Add metadata
            data['_time'] = time
            data['_session_key'] = session_key
            data['_meeting_key'] = meeting_key
            if session_time:
                data['_session_time'] = session_time

            parsed[collection_id].append(data)

    return dict(parsed)

ingestor/parsing.py
br-g/openf1
from typing import Union, Optional, Dict, Iterator
import datetime
import json
import os
import pytz
from google.oauth2.service_account import Credentials
from google.cloud import storage
import fastf1

# Disable FastF1 caching
fastf1.Cache.set_disabled() 

storage_client = storage.Client(
    credentials=Credentials.from_service_account_info(json.loads(os.environ['GOOGLE_CREDENTIALS']))
)


def upload_blob(bucket_name: str, source_file_name: str, destination_blob_name: str):
    """Uploads a file to a GCS bucket."""
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)


def to_datetime(x: Union[str, datetime.datetime]) -> Optional[datetime.datetime]:
    """This function is taken from FastF1: https://github.com/theOehrly/Fast-F1/blob/317bacf8c61038d7e8d0f48165330167702b349f/fastf1/utils.py#L178
    
    Fast datetime object creation from a date string.

    Permissible string formats:

        For example '2020-12-13T13:27:15.320000Z' with:

            - optional milliseconds and microseconds with
              arbitrary precision (1 to 6 digits)
            - with optional trailing letter 'Z'

        Examples of valid formats:

            - `2020-12-13T13:27:15.320000`
            - `2020-12-13T13:27:15.32Z`
            - `2020-12-13T13:27:15`

    Args:
        x: timestamp
    """
    if isinstance(x, str):
        try:
            date, time = x.strip('Z').split('T')
            year, month, day = date.split('-')
            hours, minutes, seconds = time.split(':')
            if '.' in seconds:
                seconds, msus = seconds.split('.')
                if len(msus) < 6:
                    msus = msus + '0' * (6 - len(msus))
                elif len(msus) > 6:
                    msus = msus[0:6]
            else:
                msus = 0

            return datetime.datetime(
                int(year), int(month), int(day), int(hours),
                int(minutes), int(seconds), int(msus)
            )

        except Exception as exc:
            #logger.debug(f"Failed to parse datetime string '{x}'")
            return None

    elif isinstance(x, datetime.datetime):
        return x

    else:
        return None


def to_timedelta(x: Union[str, datetime.timedelta]) -> Optional[datetime.timedelta]:
    """This function is taken from FastF1: https://github.com/theOehrly/Fast-F1/blob/317bacf8c61038d7e8d0f48165330167702b349f/fastf1/utils.py#L120
    
    Fast timedelta object creation from a time string

    Permissible string formats:

        For example: `13:24:46.320215` with:

            - optional hours and minutes
            - optional microseconds and milliseconds with
              arbitrary precision (1 to 6 digits)

        Examples of valid formats:

            - `24.3564` (seconds + milli/microseconds)
            - `36:54` (minutes + seconds)
            - `8:45:46` (hours, minutes, seconds)

    Args:
        x: timestamp
    """
    # this is faster than using pd.timedelta on a string
    if isinstance(x, str) and len(x):
        try:
            hours, minutes = 0, 0
            hms = x.split(':')
            if len(hms) == 3:
                hours, minutes, seconds = hms
            elif len(hms) == 2:
                minutes, seconds = hms
            else:
                seconds = hms[0]

            if '.' in seconds:
                seconds, msus = seconds.split('.')
                if len(msus) < 6:
                    msus = msus + '0' * (6 - len(msus))
                elif len(msus) > 6:
                    msus = msus[0:6]
            else:
                msus = 0

            return datetime.timedelta(
                hours=int(hours), minutes=int(minutes),
                seconds=int(seconds), microseconds=int(msus)
            )

        except Exception as exc:
            print(f"Failed to parse timedelta string '{x}'",
                          exc_info=exc)
            return None

    elif isinstance(x, datetime.timedelta):
        return x

    else:
        return None


def get_shedule(year: int) -> Iterator[Dict]:
    """Yields the schedule for the year (past and upcoming events), using FastF1"""
    for _, row in fastf1.get_event_schedule(year).iterrows():
        for i in range(1, 10):
            if f'Session{i}' not in row or row[f'Session{i}'] is None:
                continue

            yield {
                'RoundNumber': row['RoundNumber'],
                'Country': row['Country'],
                'Location': row['Location'],
                'OfficialEventName': row['OfficialEventName'],
                'Session': row[f'Session{i}'],
                'SessionDateUtc': row[f'Session{i}DateUtc'].to_pydatetime().replace(tzinfo=pytz.UTC),
            }

ingestor/util.py
br-g/openf1
"""Base class for API methods."""

from typing import List, Set, Dict, Iterable, Optional
from abc import ABC, abstractmethod
from filters import filter_results


class staticproperty(object):
    def __init__(self, getter):
        self.getter = getter

    def __get__(self, instance, owner):
        return self.getter(owner)


class BaseMethod(ABC):
    @staticproperty
    @abstractmethod
    def description(cls) -> str:
        pass

    @staticproperty
    def additional_info(cls) -> str:
        return ''

    @staticproperty
    @abstractmethod
    def example_filters(cls) -> List[str]:
        pass

    @staticproperty
    @abstractmethod
    def example_response(cls) -> List[Dict]:
        pass

    @staticproperty
    def attributes(cls) -> Optional[Set[str]]:
        return set(cls.example_response[0].keys()) if cls.example_response else None

    @classmethod
    @abstractmethod
    def _query(cls, filters: Dict[str, List[Dict]]) -> Iterable[Dict]:
        pass

    @classmethod
    def _process(cls, docs: Iterable[Dict]) -> Iterable[Dict]:
        return docs

    @classmethod
    def query_process_filter(cls, filters: Dict[str, List[Dict]]) -> List[Dict]:
        res = list(cls._process(list(cls._query(filters))))
        res = filter_results(res, filters=filters)

        attributes = cls.attributes
        if attributes:
            # Remove unwanted attributes
            res = [{k: v for k, v in e.items() if k in attributes} for e in res]
            
            # Add missing attributes
            add_missing_attributes(res, attributes=attributes)

        return res


def add_missing_attributes(docs: List[Dict], attributes: List[str]):
    """Fills missing attributes with `None`"""
    for doc in docs:
        for att in attributes:
            if att not in doc:
                doc[att] = None

query_api/base_method.py
br-g/openf1
from typing import List, Dict, Tuple, Optional, Iterator, Iterable, Optional
from copy import deepcopy
from datetime import datetime, timedelta
from pymongo import MongoClient
from filters import OPERATORS_MAP
from util import timed_cache

# Public and read only
client = MongoClient("mongodb+srv://public:nomikeyno@prod-eu.s4ponkq.mongodb.net")
db = client['raw']


def _filter_and_rename(docs: Iterable[Dict], mapping: Dict[str, str]) -> Iterator[Dict]:
    """Filters and renames attributes in retrieved MongoDB documents. Keeps only
       attributes present in `mapping`
    """
    mapping_rev = {v: k for k, v in mapping.items()}
    for doc in docs:
        res = {}
        for k, v in doc.items():
            if k in mapping_rev:
                res[mapping_rev[k]] = v
        yield res


def _flatten_doc(doc: Dict) -> Dict:
    """Flattens a nested dictionary structure by converting it into a single-level dictionary
       where keys from nested dictionaries are concatenated with a dot ('.') separator
    """
    res = {}
    for key, val in doc.items():
        if isinstance(val, dict):
            for _key, _val in _flatten_doc(val).items():
                res[f'{key}.{_key}'] = _val
        else:
            res[key] = val
    return res


def query_mongo(collection: str,
                mapping: Dict[str, Optional[str]],
                filters_user: Optional[Dict[str, List[Dict]]] = None,
                filters_other: Optional[Dict[str, List[Dict]]] = None,
                sort_by: Optional[List[Tuple[str, int]]] = None) -> List[Dict]:
    """Queries a MongoDB collection. Renames, filters and sorts"""
    # Rename using provided `mapping`
    filters_user = {mapping[k]: v for k, v in filters_user.items()} if filters_user else {}
    filters_other = {mapping[k] if k in mapping else k: v for k, v in filters_other.items()} \
                    if filters_other else {}
    sort_by = [(mapping[k] if k in mapping else k, v) for k, v in sort_by] \
              if sort_by else None
    
    # Turn `filters_user` into mongoDB filters
    filters_user_mongo = {}
    if filters_user:
        for key, vals in filters_user.items():
            filters_user_mongo[key] = {OPERATORS_MAP[e['op']]['mongo']: e['right'] for e in vals}
    
    # Combine filters
    filters_combined = deepcopy(filters_user_mongo)
    for left, vals in filters_other.items():
        if left not in filters_combined:
            filters_combined[left] = {}
        for op, right in vals.items():
            assert op not in filters_combined[left]
            filters_combined[left][op] = right

    # Query MongoDB
    docs = db[collection].find(filters_combined)

    # Sort
    if sort_by:
        docs = docs.sort(sort_by)

    # Postprocess
    docs = [_flatten_doc(e) for e in docs]
    docs = list(_filter_and_rename(docs, mapping=mapping))

    return docs


def date_to_session_key(date: datetime) -> Optional[int]:
    """If the date is during a session, returns the key of the session.
       Otherwise returns `None`
    """
    collection = db['TimingData-Lines']
    documents = list(collection.find({'_time': {
        '$gte': date - timedelta(minutes=5),
        '$lte': date + timedelta(minutes=5)
    }}).limit(1))

    if documents:
        return documents[0]['_session_key']
    else:
        return None


@timed_cache(60)  # Cache the output for 1 minute
def _get_latest_session_info() -> int:
    docs = list(db['SessionInfo'].find().sort('StartDate', -1).limit(1))
    if docs:
        return docs[0]
    else:
        raise SystemError('Could not determine the key of the latest meeting')

def get_latest_meeting_key() -> int:
    return _get_latest_session_info()['_meeting_key']

def get_latest_session_key() -> int:
    return _get_latest_session_info()['_session_key']

query_api/db.py
br-g/openf1
"""Helper functions and assets to filter results based on parameters provided by
   the user in the URL.
"""

from typing import Dict, List, Iterator, Optional
import operator
from datetime import datetime, timedelta
from collections import defaultdict
from util import has_time_info, try_cast


# Operators allowed in the URL to filter results
OPERATORS_MAP = {
    '>=': {
        'mongo': '$gte',
        'python': operator.ge,
    },
    '<=': {
        'mongo': '$lte',
        'python': operator.le,
    },
    '=': {
        'mongo': '$eq',
        'python': operator.eq,
    },
    '>': {
        'mongo': '$gt',
        'python': operator.gt,
    },
    '<': {
        'mongo': '$lt',
        'python': operator.lt,
    },
}


def _adjust_time(filter_: Dict) -> Iterator[Dict]:
    """Adjusts time filter when time info is not specified, to keep things consistent"""
    if filter_['op'] == '>':
        yield {
            'left': filter_['left'],
            'op': '>=',
            'right': filter_['right'] + timedelta(days=1),
        }
    if filter_['op'] == '<':
        yield filter_
    if filter_['op'] in {'=', '>='}:
        yield {
            'left': filter_['left'],
            'op': '>=',
            'right': filter_['right'],
        }
    if filter_['op'] in {'=', '<='}:
        yield {
            'left': filter_['left'],
            'op': '<',
            'right': filter_['right'] + timedelta(days=1),
        }


def _parse_query_filter(filter_: str) -> Iterator[Dict]:
    """Splits the filter into 3 parts: 'left', 'op' and 'right'"""
    # Prevent circular import
    from db import get_latest_meeting_key, get_latest_session_key

    for op in OPERATORS_MAP:
        if op in filter_:
            loc = filter_.find(op)
            parsed = {
                'left': filter_[:loc].lower(),
                'op': op,
                'right': filter_[loc+len(op):],
            }
            assert parsed['right'], f'Invalid query filter: `{filter_}`.'
            _right = parsed['right']
            parsed['right'] = try_cast(parsed['right'])

            # Handle value `latest` for meeting_key and session_key attributes
            if parsed['left'] == 'meeting_key' and parsed['right'] == 'latest':
                parsed['right'] = get_latest_meeting_key()
            if parsed['left'] == 'session_key' and parsed['right'] == 'latest':
                parsed['right'] = get_latest_session_key()

            # Handle time filters without time info
            if isinstance(parsed['right'], datetime) and not has_time_info(_right):
                yield from _adjust_time(parsed)
            else:
                yield parsed
            return

    raise ValueError(f'Operator not found in `{filter_}`.')

def parse_query_filters(filters: List[str]) -> Dict:
    """Parses each filter and groups them by attribute"""
    res = defaultdict(list)
    for filter_ in filters:
        for r in _parse_query_filter(filter_=filter_):
            res[r['left']].append({'op': r['op'], 'right': r['right']})
    return dict(res)


def _should_be_filtered(result: Dict, filters: Dict[str, List[Dict]]) -> Optional[Dict]:
    """Returns whether the `result` should be filtered out according to the `filters`"""
    for key, filters in filters.items():
        if result[key] is None:
            return True
        for filter_ in filters:
            if not OPERATORS_MAP[filter_['op']]['python'](result[key], filter_['right']):
                return True
    return False

def filter_results(results: List[Dict], filters: Dict[str, List[Dict]]) -> List[Dict]:
    """Filters `results` based on the provided `filters`"""
    return [e for e in results if not _should_be_filtered(e, filters)]

query_api/filters.py
br-g/openf1
import sys
from loguru import logger

# Initialize logging for Cloud Run
logger.remove()
logger.add(sys.stdout, level='INFO')

query_api/logger.py
br-g/openf1
from typing import List, Dict, Union
import importlib
from functools import lru_cache
import re
import csv
import requests
import io
import traceback
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import PlainTextResponse, HTMLResponse, Response
from fastapi.middleware.cors import CORSMiddleware
from logger import logger
from filters import parse_query_filters


app = FastAPI()

# CORS middleware settings
# There are pretty much no security risks here as the app is only reading from a public database.
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],  # Allows all origins
    allow_credentials=True,
    allow_methods=['*'],  # Allows all methods
    allow_headers=['*'],  # Allows all headers
)


@lru_cache(maxsize=1)
def fetch_favicon():
    favicon_url = 'https://storage.googleapis.com/openf1-public/images/favicon.png'
    response = requests.get(favicon_url)
    if response.status_code == 200:
        return response.content
    else:
        return None


def _results_to_csv(results: List[Dict], method: type, filename: str) -> PlainTextResponse:
    """Converts results to CSV"""
    output = io.StringIO()
    csv_writer = csv.DictWriter(output, fieldnames=sorted(list(method.attributes)))
    csv_writer.writeheader()

    for row in results:
        csv_writer.writerow(row)
    csv_data = output.getvalue()

    headers = {
        'Content-Disposition': f'attachment; filename={filename}',
    }
    return PlainTextResponse(content=csv_data, media_type='text/csv', headers=headers)


def _get_favicon() -> Response:
    image_content = fetch_favicon()
    if image_content:
        return Response(content=image_content, media_type='image/png')
    else:
        raise HTTPException(status_code=404, detail='Favicon not found')


def _process_api_query(request: Request, path: str) -> Union[List, Response]:
    # Log request
    logger.info(f'Path: {path} | Headers: {dict(request.headers)} | Query Parameters: {dict(request.query_params)}')

    # Parse filters from URL
    filters = []
    for key in request.query_params.keys():
        for value in request.query_params.getlist(key):
            if value:
                filters.append(f'{key}={value}')
            else:
                filters.append(key)

            # Handle the '+' character used to set timezone
            if 'date' in filters[-1] and re.search(r' \d{2}:\d{2}$', filters[-1]):
                filters[-1] = '+'.join(filters[-1].rsplit(' ', 1))
    filters = parse_query_filters(filters=filters)

    # Determine output format - JSON (default) or CSV
    if 'csv' in filters:
        assert len(filters['csv']) == 1
        assert filters['csv'][0]['op'] == '=', f'Invalid query parameter `{filters["csv"][0]}`'
        use_csv = filters['csv'][0]['right']
        del filters['csv']
    else:
        use_csv = False

    # Import method
    path_components = path.split('/')
    assert len(path_components) == 2, f'Invalid method URL: `{path}`'
    method_name = path_components[1].lower()
    module_name = f'methods.{path_components[0]}.{method_name}'
    try:
        module = importlib.import_module(module_name)
        method = getattr(module, 'Method')
    except:
        raise ModuleNotFoundError(f'Method `{path}` does not exist')
    
    # Check filters
    for att in filters:
        if method.attributes and att not in method.attributes:
            raise ValueError(f'Unknow attribute `{att}`')

    # Process query
    results = list(method.query_process_filter(filters))

    # Return results
    if not use_csv:
        return results
    else:
        return _results_to_csv(
            results,
            method=method,
            filename=f'{method_name}.csv'
        )


@app.api_route("/{path:path}", methods=['GET', 'POST'])
async def catch_all(request: Request, path: str):
    try:
        if path == 'favicon.ico':
            return _get_favicon()
        else:
            return _process_api_query(request, path)

    except Exception as e:
        stack_trace = traceback.format_exc()
        error_msg = f'<h1>An error occurred</h1><pre>{stack_trace}</pre>'
        stack_trace = str(stack_trace).replace('\n', ' - ')
        logger.error(f'Path: {path} | Headers: {dict(request.headers)} | Query Parameters: {dict(request.query_params)}'
                     f' | Stack trace: {stack_trace}')
        return HTMLResponse(content=error_msg, status_code=500)

query_api/main.py
br-g/openf1
from typing import List, Dict, Iterator
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return 'Some data about each car, at a sample rate of about 3.7 Hz.'
    
    @staticproperty
    def additional_info(cls) -> str:
        return '''
            Below is a table that correlates DRS values to its supposed interpretation
            (from <a href="https://github.com/theOehrly/Fast-F1/blob/317bacf8c61038d7e8d0f48165330167702b349f/fastf1/_api.py#L863" target="_blank">FastF1</a>). 

            <table id="drs_values">
                <thead>
                    <tr>
                    <th>DRS value</th>
                    <th>Interpretation</th>
                    </tr>
                </thead>
                <tbody>
                    <tr>
                    <td>0</td>
                    <td>DRS off</td>
                    </tr>
                    <tr>
                    <td>1</td>
                    <td>DRS off</td>
                    </tr>
                    <tr>
                    <td>2</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td>3</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td>8</td>
                    <td>Detected, eligible once in activation zone</td>
                    </tr>
                    <tr>
                    <td>9</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td>10</td>
                    <td>DRS on</td>
                    </tr>
                    <tr>
                    <td>12</td>
                    <td>DRS on</td>
                    </tr>
                    <tr>
                    <td>14</td>
                    <td>DRS on</td>
                    </tr>
                </tbody>
            </table>
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'driver_number=55',
            'session_key=9159',
            'speed>=315',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'driver_number': 55,
                'rpm': 11141,
                'speed': 315,
                'n_gear': 8,
                'throttle': 99,
                'brake': 0,
                'drs': 12,
                'date': '2023-09-15T13:08:19.923000+00:00',
                'session_key': 9159,
                'meeting_key': 1219
            },
            {
                'driver_number': 55,
                'rpm': 11023,
                'speed': 315,
                'n_gear': 8,
                'throttle': 57,
                'brake': 100,
                'drs': 8,
                'date': '2023-09-15T13:35:41.808000+00:00',
                'session_key': 9159,
                'meeting_key': 1219
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'meeting_key': '_meeting_key',
            'session_key': '_session_key',
            'driver_number': '_val._val._key',
            'date': '_val._Utc_',
            'rpm': '_val._val.ch_0',
            'speed': '_val._val.ch_2',
            'n_gear': '_val._val.ch_3',
            'throttle': '_val._val.ch_4',
            'drs': '_val._val.ch_45',
            'brake': '_val._val.ch_5',
        }
        docs = list(query_mongo(
            collection='CarData.z-Entries-Cars',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))

        return docs

query_api/methods/v1/car_data.py
br-g/openf1
from typing import List, Dict, Iterator
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import group_dicts_by


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return 'Provides information about drivers for each session.'

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'driver_number=1',
            'session_key=9158',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'driver_number': 1,
                'broadcast_name': 'M VERSTAPPEN',
                'full_name': 'Max VERSTAPPEN',
                'name_acronym': 'VER',
                'team_name': 'Red Bull Racing',
                'team_colour': '3671C6',
                'first_name': 'Max',
                'last_name': 'Verstappen',
                'headshot_url': 'https://www.formula1.com/content/dam/fom-website/drivers/M/MAXVER01_Max_Verstappen/maxver01.png.transform/1col/image.png',
                'country_code': 'NED',
                'session_key': 9158,
                'meeting_key': 1219
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'broadcast_name': 'BroadcastName',
            'country_code': 'CountryCode',
            'first_name': 'FirstName',
            'full_name': 'FullName',
            'headshot_url': 'HeadshotUrl',
            'last_name': 'LastName',
            'driver_number': 'RacingNumber',
            'team_colour': 'TeamColour',
            'team_name': 'TeamName',
            'name_acronym': 'Tla',
        }
        return query_mongo(
            collection='DriverList',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
        )

    @staticmethod
    def _process(docs: List[Dict]) -> Iterator[Dict]:
        docs = [e for e in docs if 'driver_number' in e]
        if not docs:
            return

        # If there are multiple documents per driver, keep only the last one
        groups = group_dicts_by(docs, attributes=['meeting_key', 'session_key', 'driver_number'])
        for _, group in sorted(list(groups.items())):
            yield group[-1]

query_api/methods/v1/drivers.py
br-g/openf1
from typing import List, Dict, Iterator, Union, Optional
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Fetches real-time interval data between drivers and their gap to the race leader.    
            Available during races only, with updates approximately every 4 seconds.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9165',
            'interval<0.005',
        ]
    
    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'gap_to_leader': 41.019,
                'interval': 0.003,
                'driver_number': 1,
                'date': '2023-09-17T13:31:02.395000+00:00',
                'session_key': 9165,
                'meeting_key': 1219
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'date': '_time',
            'driver_number': '_key',
            'gap_to_leader': 'Gap',
            'interval': 'Interval',
        }
        docs = list(query_mongo(
            collection='DriverRaceInfo',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        docs = [e for e in docs if e.get('interval') or e.get('gap_to_leader')]
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))
        return docs

    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        for doc in docs:
            doc['gap_to_leader'] = _parse_time_delta(doc.get('gap_to_leader'))
            doc['interval'] = _parse_time_delta(doc.get('interval'))
        return docs


def _parse_time_delta(time_delta: Optional[Union[str, float]]) -> Optional[Union[float, str]]:
    if not time_delta:
        return None

    # Handle leader
    if str(time_delta).startswith('LAP'):
        return None
    
    # Handle case where `time_delta` >= 60 seconds or `time_delta` >= 1 lap
    if str(time_delta).startswith('+'):
        if 'LAP' in str(time_delta):  # lapped car
            return time_delta
        else:  # `time_delta` >= 60 seconds
            minutes, seconds = map(float, time_delta.split(':'))
            return round(minutes * 60 + seconds, 3)
    
    else:
        assert isinstance(time_delta, float), \
            f'Expected float type. Found {type(time_delta)} (value: `{time_delta}`)'
        return time_delta

query_api/methods/v1/intervals.py
br-g/openf1
from typing import List, Dict, Iterator
from copy import deepcopy
from collections import defaultdict
from datetime import timedelta
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo, date_to_session_key
from util import do_try, to_timedelta, group_dicts_by


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides detailed information about individual laps.
        '''
    
    @staticproperty
    def additional_info(cls) -> str:
        return '''
            Below is a table that correlates segment values to the colors displayed on TV broadcasts.    

            <table id="segment_mapping">
                <thead>
                    <tr>
                    <th>Value</th>
                    <th>Color</th>
                    </tr>
                </thead>
                <tbody>
                    <tr>
                    <td>0</td>
                    <td>not available</td>
                    </tr>
                    <tr>
                    <td><span style="color: #fdde00;">2048</span></td>
                    <td><span style="color: #fdde00;">yellow</span></td>
                    </tr>
                    <tr>
                    <td><span style="color: #4bdd49;">2049</span></td>
                    <td><span style="color: #4bdd49;">green</span></td>
                    </tr>
                    <tr>
                    <td>2050</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td><span style="color: #c92cd5;">2051</span></td>
                    <td><span style="color: #c92cd5;">purple</span></td>
                    </tr>
                    <tr>
                    <td>2052</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td>2064</td>
                    <td>?</td>
                    </tr>
                    <tr>
                    <td>2068</td>
                    <td>?</td>
                    </tr>
                </tbody>
            </table>

            Segments are not available during races.
            Also, The segment values may not always align perfectly with the colors shown on TV, for unknown reasons.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9161',
            'driver_number=63',
            'lap_number=8',
        ]
    
    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'meeting_key': 1219,
                'session_key': 9161,
                'driver_number': 63,
                'i1_speed': 307,
                'i2_speed': 277,
                'st_speed': 298,
                'date_start': '2023-09-16T13:59:07.606000+00:00',
                'lap_duration': 91.743,
                'is_pit_out_lap': False,
                'duration_sector_1': 26.966,
                'duration_sector_2': 38.657,
                'duration_sector_3': 26.12,
                'segments_sector_1': [2049, 2049, 2049, 2051, 2049, 2051, 2049, 2049],
                'segments_sector_2': [2049, 2049, 2049, 2049, 2049, 2049, 2049, 2049],
                'segments_sector_3': [2048, 2048, 2048, 2048, 2048, 2064, 2064, 2064],
                'lap_number': 8
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        query_filters = _get_query_filters(filters)
        yield from _query_raw_lap_data(query_filters)
        yield from _query_raw_sector_data(query_filters)
        yield from _query_raw_segment_data(query_filters)

    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        docs = sorted(docs, key=lambda x: x['date'])
        laps = list(_process_laps(docs))
        laps = sorted(laps, key=lambda x: x['_tmp_date_start'])
        return laps


def _query_raw_lap_data(filters: Dict[str, List[Dict]]) -> List[Dict]:
    ATT_MAPPING = {
        'date': '_time',
        'meeting_key': '_meeting_key',
        'session_key': '_session_key',
        'driver_number': '_val._key',
        'lap_duration': '_val.LastLapTime.Value',
        'lap_number': '_val.NumberOfLaps',
        'is_pit_out_lap': '_val.PitOut',
        'i1_speed': '_val.Speeds.I1.Value',
        'i2_speed': '_val.Speeds.I2.Value',
        'st_speed': '_val.Speeds.ST.Value',
    }
    docs = list(query_mongo(
        collection='TimingData-Lines',
        mapping=ATT_MAPPING,
        filters_user=filters,
    ))
    for doc in docs:
        doc['date'] = doc['date'].replace(tzinfo=tzutc())
        doc['lap_duration'] = do_try(lambda: to_timedelta(doc['lap_duration']).total_seconds())
    return docs


def _query_raw_sector_data(filters: Dict[str, List[Dict]]) -> List[Dict]:
    ATT_MAPPING = {
        'date': '_time',
        'meeting_key': '_meeting_key',
        'session_key': '_session_key',
        'driver_number': '_val.__key_',
        'sector_duration': '_val._val.Value',
        'sector_idx': '_val._val._key',
    }
    docs = list(query_mongo(
        collection='TimingData-Lines-Sectors',
        mapping=ATT_MAPPING,
        filters_user=filters,
    ))
    for doc in docs:
        doc['date'] = doc['date'].replace(tzinfo=tzutc())
        if 'sector_duration' in doc and 'sector_idx' in doc:
            doc[f'duration_sector_{doc["sector_idx"]+1}'] = doc['sector_duration']
    return docs


def _query_raw_segment_data(filters: Dict[str, List[Dict]]) -> List[Dict]:
    ATT_MAPPING = {
        'date': '_time',
        'meeting_key': '_meeting_key',
        'session_key': '_session_key',
        'driver_number': '_val.__key_',
        'sector_idx': '_val._val.__key_',
        'segment_idx': '_val._val._val._key',
        'segment_status': '_val._val._val.Status',
    }
    docs = list(query_mongo(
        collection='TimingData-Lines-Sectors-Segments',
        mapping=ATT_MAPPING,
        filters_user=filters,
    ))
    for doc in docs:
        doc['date'] = doc['date'].replace(tzinfo=tzutc())

        # Group segment information
        try:
            doc['segment'] = {
                'sector_idx': doc['sector_idx'],
                'idx': doc['segment_idx'],
                'status': doc['segment_status'],
            }
            del doc['sector_idx']
            del doc['segment_idx']
            del doc['segment_status']
        except KeyError:
            pass

    return docs


def _get_query_filters(filters: Dict) -> Dict:
    """We need to process full sessions for maintaining accuracy.
       Filters are adjusted to match this constraint."""
    filters = deepcopy(filters)

    if 'date_start' in filters:
        for filter_ in filters['date_start']:
            session_key = date_to_session_key(filter_['right'])
            if session_key:
                if 'session_key' not in filters:
                    filters['session_key'] = []
                filters['session_key'].append({
                    'op': filter_['op'] + ('=' if '=' not in filter_['op'] else ''),
                    'right': session_key,
                })
            else:
                if 'date' not in filters:
                    filters['date'] = []
                filters['date'].append(filter_)
            
            if filter_['op'] in {'=', '<', '<='}:
                filter_['right'] += timedelta(minutes=5)

        del filters['date_start']

    return {k: v for k, v in filters.items() if k in {
        'date',
        'meeting_key',
        'session_key',
        'driver_number',
    }}


def _group_docs_by_driver_session_lap(docs: List[Dict]) -> Iterator[List[Dict]]:
    """To make sense of data, it is necessary to group it by driver, session and lap."""
    docs_by_driver = group_dicts_by(docs, ['session_key', 'driver_number'])
    for docs in docs_by_driver.values():
        docs_by_lap = [defaultdict(list)]

        for doc in docs:
            if doc.get('lap_number'):
                docs_by_lap.append(defaultdict(list))
            for key, val in doc.items():
                if not val:
                    continue
                
                # If data expected at the end of a lap is received at the begining of a lap,
                # assign the data to the previous lap.
                is_lap_start = 'date' not in docs_by_lap[-1] or (doc['date'] - docs_by_lap[-1]['date'][0]).total_seconds() < 10
                if is_lap_start and key == 'lap_duration':
                    if len(docs_by_lap) >= 2:
                        docs_by_lap[-2][key].append(val)
                elif is_lap_start and key.startswith('duration_sector_'):
                    if len(docs_by_lap) >= 2:
                        docs_by_lap[-2][key].append(val)
                elif key == 'segment' and val['sector_idx'] == 2:
                    if len(docs_by_lap) >= 2:
                        docs_by_lap[-2][key].append(val)
                else:
                    docs_by_lap[-1][key].append(val)

        yield docs_by_lap


def _process_lap(docs_values: Dict[str, List]) -> Dict:
    """Determines lap information given all the raw data received for this lap"""
    lap = {
        'meeting_key': do_try(lambda: docs_values['meeting_key'][0]),
        'session_key': do_try(lambda: docs_values['session_key'][0]),
        'driver_number': do_try(lambda: docs_values['driver_number'][0]),
        'i1_speed': do_try(lambda: docs_values['i1_speed'][-1]),
        'i2_speed': do_try(lambda: docs_values['i2_speed'][-1]),
        'st_speed': do_try(lambda: docs_values['st_speed'][-1]),
        'date_start': do_try(lambda: docs_values['date'][0]),
        'lap_duration': do_try(lambda: docs_values['lap_duration'][-1]),
        'is_pit_out_lap': do_try(lambda: docs_values['is_pit_out_lap'][0]),
    }
    lap['is_pit_out_lap'] = True if lap['is_pit_out_lap'] else False

    for i in range(1, 4):
        lap[f'duration_sector_{i}'] = do_try(lambda: docs_values[f'duration_sector_{i}'][-1])
    
    # Fill in missing lap duration from sector duration, if possible
    sector_dur_sum = do_try(lambda: abs(sum(lap[f'duration_sector_{i}'] for i in range(1, 4))))
    if sector_dur_sum and not lap['lap_duration']:
        lap['lap_duration'] = round(sector_dur_sum, 3)
    
    # Fill in missing sector duration from lap duration, if possible
    for i in range(1, 4):
        if lap[f'duration_sector_{i}'] is None:
            lap[f'duration_sector_{i}'] = 0.
            other_sectors_sum = do_try(lambda: sum(lap[f'duration_sector_{i}'] for i in range(1, 4)))
            lap[f'duration_sector_{i}'] = do_try(lambda: round(lap['lap_duration'] - other_sectors_sum, 3))
    
    # Segments
    for i in range(1, 4):
        sector_segs = [e for e in docs_values['segment'] if e['sector_idx'] == i-1]
        indices = [e['idx'] for e in sector_segs if e['idx'] is not None]
        lap[f'segments_sector_{i}'] = [None] * (max(indices)+1) if indices else []
        for seg in sector_segs:
            if seg['idx'] is None:
                continue
            lap[f'segments_sector_{i}'][seg['idx']] = seg['status']

    return lap


def _process_laps(docs: List[Dict]) -> Iterator[Dict]:
    """Computes lap information from raw data."""
    for docs_group in _group_docs_by_driver_session_lap(docs):
        laps = [_process_lap(e) for e in docs_group]

        # Remove first and last lap if they don't look good
        ATTS = {
            'i1_speed',
            'i2_speed',
            'st_speed',
        }
        if laps and all(laps[0][e] is None for e in ATTS):
            laps = laps[1:]
        if laps and all(laps[-1][e] is None for e in ATTS):
            laps = laps[:-1]

        # Add lap number (this information in raw data is not reliable)
        for i, lap in enumerate(laps):
            lap['lap_number'] = i + 1
        
        # Remove the `start_date` value for the first lap if it looks incorrect
        for lap in laps:
            lap['_tmp_date_start'] = lap['date_start']
        if len(laps) >= 1:
            delta = do_try(lambda: laps[1]['date_start'] - laps[0]['date_start'])
            if delta is None or not timedelta(seconds=45) < delta < timedelta(minutes=20):
                laps[0]['date_start'] = None
        
        yield from laps

query_api/methods/v1/laps.py
br-g/openf1
from typing import List, Dict, Iterator
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            The approximate location of the cars on the circuit, at a sample rate of about 3.7 Hz.    
            Useful for gauging their progress along the track, but lacks details about lateral placement — i.e. whether
            the car is on the left or right side of the track. The origin point (0, 0, 0) appears to be arbitrary
            and not tied to any specific location on the track.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9161',
            'driver_number=81',
            'date>2023-09-16T13:03:35.200',
            'date<2023-09-16T13:03:35.800',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'x': 567,
                'y': 3195,
                'z': 187,
                'driver_number': 81,
                'date': '2023-09-16T13:03:35.292000+00:00',
                'session_key': 9161,
                'meeting_key': 1219
            },
            {
                'x': 489,
                'y': 3403,
                'z': 186,
                'driver_number': 81,
                'date': '2023-09-16T13:03:35.752000+00:00',
                'session_key': 9161,
                'meeting_key': 1219
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'meeting_key': '_meeting_key',
            'session_key': '_session_key',
            'driver_number': '_val._val._key',
            'date': '_val._Timestamp_',
            'x': '_val._val.X',
            'y': '_val._val.Y',
            'z': '_val._val.Z',
        }
        docs = list(query_mongo(
            collection='Position.z-Position-Entries',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))

        return docs

query_api/methods/v1/location.py
br-g/openf1
from typing import List, Dict, Iterator
from collections import defaultdict
from datetime import datetime
import pytz
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import add_timezone_info, do_try, group_dicts_by


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides information about meetings.    
            A meeting refers to a Grand Prix or testing weekend and usually includes multiple sessions (practice, qualifying, race, ...).
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'year=2023',
            'country_name=Singapore',
        ]
    
    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'meeting_name': 'Singapore Grand Prix',
                'meeting_official_name': 'FORMULA 1 SINGAPORE AIRLINES SINGAPORE GRAND PRIX 2023',
                'location': 'Marina Bay',
                'country_key': 157,
                'country_code': 'SGP',
                'country_name': 'Singapore',
                'circuit_key': 61,
                'circuit_short_name': 'Singapore',
                'date_start': '2023-09-15T09:30:00+00:00',
                'gmt_offset': '08:00:00',
                'meeting_key': 1219,
                'year': 2023
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        filters = _replace_year_filter(filters)
        MAPPING = {
            'circuit_key': 'Meeting.Circuit.Key',
            'circuit_short_name': 'Meeting.Circuit.ShortName',
            'meeting_key': '_meeting_key',
            'meeting_code': 'Meeting.Country.Code',
            'location': 'Meeting.Location',
            'country_key': 'Meeting.Country.Key',
            'country_code': 'Meeting.Country.Code',
            'country_name': 'Meeting.Country.Name',
            'meeting_name': 'Meeting.Name',
            'meeting_official_name': 'Meeting.OfficialName',
            'gmt_offset': 'GmtOffset',
            'date_start': 'StartDate',
            '_session_name': 'Name',
        }
        return query_mongo(
            collection='SessionInfo',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date_start', 1)],
        )
        
    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        docs = [docs[0] for docs in group_dicts_by(docs, attributes=['meeting_key']).values()]

        # Handle case where the `date_start` user filter is in the middle of a GP
        if docs and docs[0]['_session_name'].lower() != 'practice 1':
            docs = docs[1:]
        
        for doc in docs:
            doc['date_start'] = do_try(lambda: add_timezone_info(dt=doc['date_start'], gmt_offset=doc['gmt_offset']))
            doc['year'] = do_try(lambda: doc['date_start'].year)
            doc['meeting_official_name'] = doc['meeting_official_name'].strip(' ')
        return sorted(docs, key=lambda x: x['date_start'])


def _replace_year_filter(filters: Dict) -> Dict:
    """Replaces the `year` filter with a `date_start` equivalent (`year` doesn't
       exist in raw data)"""
    if 'year' in filters:
        filters = defaultdict(list, filters)
        for filter in filters['year']:
            if filter['op'] in {'>=', '>', '='}:
                filters['date_start'].append({
                    'op': '>=' if filter['op'] == '=' else filter['op'],
                    'right': datetime(filter['right'], 1, 1).replace(tzinfo=pytz.utc),
                })
            if filter['op'] in {'<=', '<', '='}:
                filters['date_start'].append({
                    'op': '<' if filter['op'] == '=' else filter['op'],
                    'right': datetime(filter['right']+1, 1, 1).replace(tzinfo=pytz.utc),
                })
        del filters['year']
    return filters

query_api/methods/v1/meetings.py
br-g/openf1
from typing import List, Dict, Iterator, Union, Optional
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides information about cars going through the pit lane.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9158',
            'pit_duration<31',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'pit_duration': 24.5,
                'lap_number': 5,
                'driver_number': 63,
                'date': '2023-09-15T09:38:23.038000+00:00',
                'session_key': 9158,
                'meeting_key': 1219
            },
            {
                'pit_duration': 30.8,
                'lap_number': 13,
                'driver_number': 81,
                'date': '2023-09-15T10:05:01.229000+00:00',
                'session_key': 9158,
                'meeting_key': 1219
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'date': '_time',
            'driver_number': '_val._key',
            'pit_duration': '_val.Duration',
            'lap_number': '_val.Lap',
        }
        docs = list(query_mongo(
            collection='PitLaneTimeCollection-PitTimes',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        docs = [e for e in docs if e.get('pit_duration')]
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))
        return docs

query_api/methods/v1/pit.py
br-g/openf1
from typing import List, Dict, Iterator
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides driver positions throughout a session, including initial
            placement and subsequent changes.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'meeting_key=1217',
            'driver_number=40',
            'position<=3',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'position': 2,
                'driver_number': 40,
                'date': '2023-08-26T09:30:47.199000+00:00',
                'session_key': 9144,
                'meeting_key': 1217
            },
            {
                'position': 3,
                'driver_number': 40,
                'date': '2023-08-26T09:35:51.477000+00:00',
                'session_key': 9144,
                'meeting_key': 1217
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'driver_number': '_val._key',
            'date': '_time',
            'position': '_val.Line',
        }
        docs = list(query_mongo(
            collection='TimingAppData-Lines',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        docs = [e for e in docs if 'position' in e]
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))
        return docs

query_api/methods/v1/position.py
br-g/openf1
from typing import List, Dict, Iterator
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides information about race control (racing incidents, flags, safety car, ...).
        '''
    
    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'flag=BLACK AND WHITE',
            'driver_number=1',
            'date>=2023-01-01',
            'date<2023-09-01',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'date': '2023-06-04T14:21:01+00:00',
                'lap_number': 59,
                'category': 'Flag',
                'flag': 'BLACK AND WHITE',
                'scope': 'Driver',
                'driver_number': 1,
                'message': 'BLACK AND WHITE FLAG FOR CAR 1 (VER) - TRACK LIMITS',
                'session_key': 9102,
                'meeting_key': 1211,
                'sector': None,
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'date': '_val.Utc',
            'category': '_val.Category',
            'flag': '_val.Flag',
            'lap_number': '_val.Lap',
            'message': '_val.Message',
            'driver_number': '_val.RacingNumber',
            'scope': '_val.Scope',
            'sector': '_val.Sector',
        }
        docs = list(query_mongo(
            collection='RaceControlMessages-Messages',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))
        return docs

query_api/methods/v1/race_control.py
br-g/openf1
from typing import List, Dict, Iterator
from collections import defaultdict
from datetime import datetime
import pytz
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import add_timezone_info, do_try, deduplicate_dicts


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides information about sessions.    
            A session refers to a distinct period of track activity during a Grand Prix or testing weekend (practice, qualifying, sprint, race, ...).
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'country_name=Belgium',
            'session_name=Sprint',
            'year=2023',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'location': 'Spa-Francorchamps',
                'country_key': 16,
                'country_code': 'BEL',
                'country_name': 'Belgium',
                'circuit_key': 7,
                'circuit_short_name': 'Spa-Francorchamps',
                'session_type': 'Race',
                'session_name': 'Sprint',
                'date_start': '2023-07-29T15:05:00+00:00',
                'date_end': '2023-07-29T15:35:00+00:00',
                'gmt_offset': '02:00:00',
                'session_key': 9140,
                'meeting_key': 1216,
                'year': 2023
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        filters = _replace_year_filter(filters)
        MAPPING = {
            'session_key': '_session_key',
            'session_name': 'Name',
            'date_start': 'StartDate',
            'date_end': 'EndDate',
            'gmt_offset': 'GmtOffset',
            'session_type': 'Type',
            'meeting_key': '_meeting_key',
            'location': 'Meeting.Location',
            'country_key': 'Meeting.Country.Key',
            'country_code': 'Meeting.Country.Code',
            'country_name': 'Meeting.Country.Name',
            'circuit_key': 'Meeting.Circuit.Key',
            'circuit_short_name': 'Meeting.Circuit.ShortName',
        }
        return query_mongo(
            collection='SessionInfo',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date_start', 1)],
        )
        
    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        for doc in docs:
            doc['date_start'] = do_try(lambda: add_timezone_info(dt=doc['date_start'], gmt_offset=doc['gmt_offset']))
            doc['date_end'] = do_try(lambda: add_timezone_info(dt=doc['date_end'], gmt_offset=doc['gmt_offset']))
            doc['year'] = do_try(lambda: doc['date_start'].year)
        docs = deduplicate_dicts(docs)
        return docs


def _replace_year_filter(filters: Dict) -> Dict:
    """Replaces the `year` filter with a `date_start` equivalent (`year` doesn't
       exist in raw data)"""
    if 'year' in filters:
        filters = defaultdict(list, filters)
        for filter in filters['year']:
            if filter['op'] in {'>=', '>', '='}:
                filters['date_start'].append({
                    'op': '>=' if filter['op'] == '=' else filter['op'],
                    'right': datetime(filter['right'], 1, 1).replace(tzinfo=pytz.utc),
                })
            if filter['op'] in {'<=', '<', '='}:
                filters['date_end'].append({
                    'op': '<' if filter['op'] == '=' else filter['op'],
                    'right': datetime(filter['right']+1, 1, 1).replace(tzinfo=pytz.utc),
                })
        del filters['year']
    return filters

query_api/methods/v1/sessions.py
br-g/openf1
from typing import List, Dict, Iterator
from collections import defaultdict
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import group_dicts_by, do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides information about individual stints.    
            A stint refers to a period of continuous driving by a driver during a session.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9165',
            'tyre_age_at_start>=3',
        ]
    
    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'meeting_key': 1219,
                'session_key': 9165,
                'stint_number': 1,
                'driver_number': 16,
                'lap_start': 1,
                'lap_end': 20,
                'compound': 'SOFT',
                'tyre_age_at_start': 3
            },
            {
                'meeting_key': 1219,
                'session_key': 9165,
                'stint_number': 3,
                'driver_number': 20,
                'lap_start': 44,
                'lap_end': 62,
                'compound': 'SOFT',
                'tyre_age_at_start': 3
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> List[Dict]:
        yield from query_mongo(
            collection='TimingAppData-Lines-Stints',
            mapping={
                'session_key': '_session_key',
                'meeting_key': '_meeting_key',
                'driver_number': '_val.__key_',
                'time': '_time',
                'tyre_age': '_val._val.TotalLaps',
                'compound': '_val._val.Compound',
                'stint_number': '_val._val._key',
            },
            filters_user={k: v for k, v in filters.items() if k in {'session_key', 'meeting_key', 'driver_number'}},
        )
        yield from query_mongo(
            collection='CurrentTyres-Tyres',
            mapping={
                'session_key': '_session_key',
                'meeting_key': '_meeting_key',
                'driver_number': '_val._key',
                'time': '_time',
                'compound': '_val.Compound',
            },
            filters_user={k: v for k, v in filters.items() if k in {'session_key', 'meeting_key', 'driver_number'}},
        )
        yield from query_mongo(
            collection='TimingData-Lines',
            mapping={
                'time': '_time',
                'meeting_key': '_meeting_key',
                'session_key': '_session_key',
                'driver_number': '_val._key',
                'lap_number': '_val.NumberOfLaps',
                'i1_speed': '_val.Speeds.I1.Value',
                'i2_speed': '_val.Speeds.I2.Value',
                'st_speed': '_val.Speeds.ST.Value',
            },
            filters_user={k: v for k, v in filters.items() if k in {'session_key', 'meeting_key', 'driver_number'}},
        )

    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        docs = sorted(docs, key=lambda x: x['time'])

        stints = []
        for _docs in group_dicts_by(docs, ['session_key', 'driver_number']).values():
            stints += list(_process_stints(docs=_docs))
        
        return sorted(stints, key=lambda x: x['time'])


def _group_docs_by_stint(docs: List[Dict]) -> Iterator[List[Dict]]:
    buf = []
    cur_stint = 0
    for doc in docs:
        if 'stint_number' in doc and doc['stint_number'] > cur_stint:
            if buf:
                yield buf
            buf = []
            cur_stint = doc['stint_number']
        buf.append(doc)

    if buf:    
        yield buf


def _get_data_by_lap(docs: List[Dict]) -> Iterator[List[Dict]]:
    buf = defaultdict(list)
    for doc in docs:
        if doc.get('lap_number'):
            yield buf
            buf = defaultdict(list)
        for key, val in doc.items():
            if val == '':
                continue
            buf[key].append(val)
    yield buf


def _count_laps(docs: List[Dict], is_first_stint: bool, is_last_stint: bool) -> int:
    """Counts number of laps in a stint"""
    n_laps = 0
    data_by_lap = list(_get_data_by_lap(docs))

    # Prevent from counting the first/last lap of a stint twice
    if not is_last_stint:
        data_by_lap = data_by_lap[:-1]

    for lap_number, lap_data in enumerate(data_by_lap):

        # Skip first and last lap of the session if they "don't look good"
        ATTS = {
            'i1_speed',
            'i2_speed',
            'st_speed',
        }
        if (is_first_stint and lap_number == 0
            and all(not lap_data[att] for att in ATTS)):
            continue
        if (is_last_stint and lap_number == len(data_by_lap)-1
            and all(not lap_data[att] for att in ATTS)):
            continue

        n_laps += 1
    
    return n_laps


def _process_stints(docs: List[Dict]) -> Iterator[Dict]:
    """Processes stints for a single session and driver"""
    docs_by_stint = list(_group_docs_by_stint(docs))

    lap_start = [1]
    for i, docs in enumerate(docs_by_stint):
        n_laps = _count_laps(
            docs=docs,
            is_first_stint=i == 0,
            is_last_stint=i == len(docs_by_stint)-1,
        )
        lap_start.append(lap_start[-1] + n_laps)

    stint_number = 1
    for i, docs in enumerate(docs_by_stint):
        if lap_start[i] == lap_start[i+1]:  # stint has not yet started
            continue

        yield {
            'meeting_key': do_try(lambda: docs[0]['meeting_key']),
            'session_key': do_try(lambda: docs[0]['session_key']),
            'stint_number': stint_number,
            'driver_number': do_try(lambda: docs[0]['driver_number']),
            'lap_start': lap_start[i],
            'lap_end': lap_start[i+1] - 1,
            'compound': do_try(lambda: [d['compound'] for d in docs if 'compound' in d][-1]),
            'tyre_age_at_start': do_try(lambda: [e.get('tyre_age') for e in docs if e.get('tyre_age') is not None][0]),
            'time': do_try(lambda: docs[0]['time']),
        }
        stint_number += 1

query_api/methods/v1/stints.py
br-g/openf1
from typing import List, Dict, Iterator, Optional
from base_method import staticproperty, BaseMethod
from db import db, query_mongo
from util import timed_cache


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return '''
            Provides a collection of radio exchanges between Formula 1 drivers and their respective teams during sessions.    
            Please note that only a limited selection of communications are included, not the complete record of radio interactions.
        '''

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'session_key=9158',
            'driver_number=11',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'date': '2023-09-15T09:40:43.005000',
                'driver_number': 11,
                'session_key': 9158,
                'meeting_key': 1219,
                'recording_url': 'https://livetiming.formula1.com/static/2023/2023-09-17_Singapore_Grand_Prix/2023-09-15_Practice_1/TeamRadio/SERPER01_11_20230915_104008.mp3'
            },
            {
                'date': '2023-09-15T10:32:47.325000',
                'driver_number': 11,
                'session_key': 9158,
                'meeting_key': 1219,
                'recording_url': 'https://livetiming.formula1.com/static/2023/2023-09-17_Singapore_Grand_Prix/2023-09-15_Practice_1/TeamRadio/SERPER01_11_20230915_113201.mp3'
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'session_key': '_session_key',
            'meeting_key': '_meeting_key',
            'driver_number': '_val.RacingNumber',
            'date': '_val.Utc',
            '_path': '_val.Path',
        }
        return query_mongo(
            collection='TeamRadio-Captures',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        )

    @staticmethod
    def _process(docs: List[Dict]) -> List[Dict]:
        for doc in docs:
            session_url = _get_session_url(doc['session_key'])
            if session_url:
                doc['recording_url'] = session_url + doc['_path']
        
        return [e for e in docs if 'recording_url' in e]


@timed_cache(300)  # Cache the output for 5 minutes
def _get_session_url(session_key: int) -> Optional[str]:
    """Returns the base URL of a specific session"""
    docs = list(db['SessionInfo'].find({'_session_key': {'$eq': session_key}}))
    if not docs:
        return None
    return 'https://livetiming.formula1.com/static/' + docs[-1]['Path']

query_api/methods/v1/team_radio.py
br-g/openf1
from typing import List, Dict, Iterator
from dateutil.tz import tzutc
from base_method import staticproperty, BaseMethod
from db import query_mongo
from util import do_try


class Method(BaseMethod):
    @staticproperty
    def description(cls) -> str:
        return 'The weather over the track, updated every minute.'

    @staticproperty
    def example_filters(cls) -> List[str]:
        return [
            'meeting_key=1208',
            'wind_direction>=130',
            'track_temperature>=52',
        ]

    @staticproperty
    def example_response(cls) -> List[Dict]:
        return [
            {
                'air_temperature': 27.8,
                'humidity': 58,
                'pressure': 1018.7,
                'rainfall': 0,
                'track_temperature': 52.5,
                'wind_direction': 136,
                'wind_speed': 2.4,
                'date': '2023-05-07T18:42:25.233000+00:00',
                'session_key': 9078,
                'meeting_key': 1208
            }
        ]

    @staticmethod
    def _query(filters: Dict[str, List[Dict]]) -> Iterator[Dict]:
        MAPPING = {
            'meeting_key': '_meeting_key',
            'session_key': '_session_key',
            'date': '_time',
            'air_temperature': 'AirTemp',
            'humidity': 'Humidity',
            'pressure': 'Pressure',
            'rainfall': 'Rainfall',
            'track_temperature': 'TrackTemp',
            'wind_direction': 'WindDirection',
            'wind_speed': 'WindSpeed',
        }
        docs = list(query_mongo(
            collection='WeatherData',
            mapping=MAPPING,
            filters_user={k: v for k, v in filters.items()},
            sort_by=[('date', 1)],
        ))
        for doc in docs:
            doc['date'] = do_try(lambda: doc['date'].replace(tzinfo=tzutc()))

        return docs

query_api/methods/v1/weather.py
br-g/openf1
"""Measure delay in real-time during a session."""

from datetime import datetime
import sys
sys.path.append('..')
from db import db


def measure():
    time_start = datetime.utcnow()
    docs = list(db['Position.z-Position-Entries'].find().sort('_time', -1).limit(1))
    assert docs
    time_end = datetime.utcnow()
    print(f'delay: {time_start - docs[0]["_time"]}')


while True:
    measure()

query_api/scripts/measure_delay.py
br-g/openf1
"""
Test each API method individually using the `example_filters` and `example_response` provided.
To run this test from the `query_api` directory: `pytest tests/integration.py`.
"""

from typing import List, Dict
import sys
from pathlib import Path
from datetime import datetime
import pytest

sys.path.append('.')
from util import get_api_methods
from filters import parse_query_filters

_METHODS_DIR = Path('./methods')


def _datetimes_to_iso(list_of_dicts: List[Dict]):
    # Convert all datetime objects to ISO format
    for dct in list_of_dicts:
        for key, value in dct.items():
            if isinstance(value, datetime):
                dct[key] = value.isoformat()


_all_methods = list(get_api_methods(_METHODS_DIR))

@pytest.mark.parametrize('url, method', _all_methods)
def test_api_method(url, method):
    print(f'Testing method: `{url}`')
    filters = parse_query_filters(method.example_filters)
    res = list(method.query_process_filter(filters))
    _datetimes_to_iso(res)
    assert res == method.example_response

query_api/tests/integration.py
br-g/openf1
from typing import Dict, List, Tuple, Iterator, Callable, Any, Union, Optional
from functools import wraps
import sys
import time
from collections import defaultdict
import importlib
from datetime import datetime, timedelta
from pathlib import Path
from dateutil.parser import parse
from dateutil.tz import tzutc
import pytz


def parse_date(date_string: str) -> datetime:
    dt = parse(date_string)
    if dt.tzinfo is None or dt.tzinfo.utcoffset(dt) is None:
        dt = dt.replace(tzinfo=tzutc())
    else:
        dt = dt.astimezone(tzutc())
    return dt


def do_try(fn: Callable, *args, **kwargs) -> Any:
    try:
        return fn(*args, **kwargs)
    except:
        return None


def add_timezone_info(dt: datetime, gmt_offset: str) -> datetime:
    h, _, _ = map(int, gmt_offset.split(':'))
    offset_tz = pytz.FixedOffset(h * 60)
    return dt.replace(tzinfo=offset_tz).astimezone(tzutc())


def to_timedelta(x: Union[str, timedelta]) \
        -> Optional[timedelta]:
    """This function is taken from FastF1: https://github.com/theOehrly/Fast-F1/blob/317bacf8c61038d7e8d0f48165330167702b349f/fastf1/utils.py#L120
    
    Fast timedelta object creation from a time string

    Permissible string formats:

        For example: `13:24:46.320215` with:

            - optional hours and minutes
            - optional microseconds and milliseconds with
              arbitrary precision (1 to 6 digits)

        Examples of valid formats:

            - `24.3564` (seconds + milli/microseconds)
            - `36:54` (minutes + seconds)
            - `8:45:46` (hours, minutes, seconds)

    Args:
        x: timestamp
    """
    # this is faster than using pd.timedelta on a string
    if isinstance(x, str) and len(x):
        try:
            hours, minutes = 0, 0
            if len(hms := x.split(':')) == 3:
                hours, minutes, seconds = hms
            elif len(hms) == 2:
                minutes, seconds = hms
            else:
                seconds = hms[0]

            if '.' in seconds:
                seconds, msus = seconds.split('.')
                if len(msus) < 6:
                    msus = msus + '0' * (6 - len(msus))
                elif len(msus) > 6:
                    msus = msus[0:6]
            else:
                msus = 0

            return timedelta(
                hours=int(hours), minutes=int(minutes),
                seconds=int(seconds), microseconds=int(msus)
            )

        except Exception as exc:
            print(f"Failed to parse timedelta string '{x}'",
                          exc_info=exc)
            return None

    elif isinstance(x, timedelta):
        return x

    else:
        return None


def group_dicts_by(dicts: List[Dict], attributes: List[str]) -> Dict[Tuple, List[Dict]]:
    """Groups dicts by a set of attributes"""
    assert attributes
    res = defaultdict(list)
    for dic in dicts:
        if any(e not in dic for e in attributes):
            continue
        key = tuple([dic[e] for e in attributes]) if len(attributes) > 1 else dic[attributes[0]]
        res[key].append(dic)
    return dict(res)


def timed_cache(expiration_time: float) -> Callable:
    """A decorator to cache the function output for a given duration.
       `expiration_time` is in seconds.
    """
    cache = {}
    timestamps = {}
    
    def decorator(func):
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Convert the arguments and keyword arguments to a single hashable key
            key = (args, frozenset(kwargs.items()))
            
            # Check if value is in cache and not expired
            if key in cache:
                if time.time() - timestamps[key] < expiration_time:
                    return cache[key]
                else:
                    # Remove the expired data
                    del cache[key]
                    del timestamps[key]

            # Compute the result and cache it
            result = func(*args, **kwargs)
            cache[key] = result
            timestamps[key] = time.time()
            
            return result
        
        return wrapper
    
    return decorator


def has_time_info(date_str: str) -> bool:
    """Returns whether the date string has time info.
       eg.
           - "2021-09-10" -> False
           - "2021-09-10T14:30:20" -> True
    """
    # Parse the string with the first default datetime
    default_datetime1 = datetime(1, 1, 1, 5, 0)  # 5:00
    dt1 = parse(date_str, default=default_datetime1)
    
    # Parse the string with the second default datetime
    default_datetime2 = datetime(1, 1, 1, 10, 0)  # 10:00
    dt2 = parse(date_str, default=default_datetime2)
    
    # Check if hours and minutes are the same
    return dt1.hour == dt2.hour


def _try_parse_boolean(s: str) -> Union[bool, str]:
    """Casts to boolean if possible"""
    if s.lower() == 'true':
        return True
    elif s.lower() == 'false':
        return False
    else:
        return s

def _try_parse_number(s: str) -> Union[int, float, str]:
    """Casts to int or float if possible"""
    try:
        return int(s)
    except ValueError:
        try:
            return float(s)
        except ValueError:
            return s

def _try_parse_date(s: str) -> Union[datetime, str]:
    """Casts to datetime if possible"""
    try:
        return parse_date(s)
    except:
        return s

def try_cast(obj: Any) -> Any:
    """Casts to boolean, float or datetime if possible"""
    if isinstance(obj, str):
        obj = _try_parse_boolean(obj)
        if not isinstance(obj, str):
            return obj
        obj = _try_parse_number(obj)
        if not isinstance(obj, str):
            return obj
        return _try_parse_date(obj)
    else:
        return obj


def _get_python_filepaths(base_dir: Path) -> Iterator[Path]:
    """Yields Python files from a given directory (and nested sub-directories)"""
    for path in base_dir.iterdir():
        if path.name[0] == '_':
            continue
        if path.is_dir():
            yield from _get_python_filepaths(path)
        elif path.suffix == '.py':
            yield path

def _get_classes_from_file(filepath: Path) -> Iterator[type]:
    """Yields classes from a Python file"""
    spec = importlib.util.spec_from_file_location(filepath.stem, filepath)
    module = importlib.util.module_from_spec(spec)
    
    sys.modules[filepath.stem] = module
    spec.loader.exec_module(module)

    for obj in vars(module).values():
        if isinstance(obj, type):
            yield obj

def get_api_methods(base_dir: Path) -> Iterator[Tuple[str, 'BaseMethod']]:
    """Yields the API methods (URL and class)"""
    # Prevent circular import
    from base_method import BaseMethod

    for filepath in _get_python_filepaths(base_dir):
        classes = [e for e in _get_classes_from_file(filepath)
                   if issubclass(e, BaseMethod) and e != BaseMethod]
        assert len(classes) <= 1, classes
        if classes:
            filepath = filepath.relative_to(base_dir)
            yield str(filepath.parent / filepath.stem), classes[0]


def deduplicate_dicts(dicts: List[Dict]) -> List[Dict]:
    seen = set()
    res = []
    for d in dicts:
        t = tuple(sorted(d.items()))
        if t not in seen:
            seen.add(t)
            res.append(d)
    return res

query_api/util.py
br-g/openf1