"""
Timing and Telemetry Data - :mod:`fastf1.core`
==============================================
The Fast-F1 core is a collection of functions and data objects for accessing
and analyzing F1 timing and telemetry data.
Data Objects
------------
All data is provided through the following data objects:
.. autosummary::
:nosignatures:
Weekend
Session
Laps
Lap
Telemetry
SessionResults
DriverResult
The :class:`Session` object is mainly used as an entry point for loading
timing data and telemetry data. The :class:`Session` can create a
:class:`Laps` object which contains all timing, track and session status
data for a whole session.
Usually you will be using :func:`get_session` to get a :class:`Session`
object.
The :class:`Laps` object holds detailed information about multiples laps.
The :class:`Lap` object holds the same information as :class:`Laps` but only
for one single lap. When selecting a single lap from a :class:`Laps` object,
an object of type :class:`Lap` will be returned.
Apart from only providing data, the :class:`Laps`, :class:`Lap` and
:class:`Telemetry` objects implement various methods for selecting and
analyzing specific parts of the data.
Functions
---------
.. autosummary::
:nosignatures:
get_session
get_round
"""
import collections
from functools import cached_property
import logging
import warnings
import numpy as np
import pandas as pd
import fastf1
from fastf1 import api, ergast
from fastf1.utils import recursive_dict_get, to_timedelta
logging.basicConfig(level=logging.INFO, style='{',
format="{module: <8} {levelname: >10} \t{message}")
D_LOOKUP = [[44, 'HAM', 'Mercedes'], [77, 'BOT', 'Mercedes'],
[55, 'SAI', 'Ferrari'], [16, 'LEC', 'Ferrari'],
[33, 'VER', 'Red Bull'], [11, 'PER', 'Red Bull'],
[3, 'RIC', 'McLaren'], [4, 'NOR', 'McLaren'],
[5, 'VET', 'Aston Martin'], [18, 'STR', 'Aston Martin'],
[14, 'ALO', 'Alpine'], [31, 'OCO', 'Alpine'],
[22, 'TSU', 'AlphaTauri'], [10, 'GAS', 'AlphaTauri'],
[47, 'MSC', 'Haas F1 Team'], [9, 'MAZ', 'Haas F1 Team'],
[7, 'RAI', 'Alfa Romeo'], [99, 'GIO', 'Alfa Romeo'],
[6, 'LAT', 'Williams'], [63, 'RUS', 'Williams']]
def get_session(*args, **kwargs):
"""
.. deprecated:: 2.2
replaced by :func:`fastf1.get_session`
"""
# TODO remove
warnings.warn("`fastf1.core.get_session` has been deprecated and will be"
"removed in a future version.\n"
"Use `fastf1.get_session` instead.", FutureWarning)
from fastf1 import events
return events.get_session(*args, **kwargs)
def get_round(year, match):
"""
.. deprecated:: 2.2
will be removed without replacement;
Use :func:`fastf1.get_event` instead to get an
:class:`~fastf1.events.Event` object which provides
information including the round number for the event.
"""
# TODO remove
warnings.warn("_func:`fastf1.core.get_round` has been deprecated and will "
"be removed without replacement in a future version.\n"
"Use :func:`fastf1.get_event` instead to get an "
":class:`~fastf1.events.Event` object which provides "
"information including the round number for the event.",
FutureWarning)
from fastf1 import events
event = events.get_event(year, match)
return event.RoundNumber
[docs]class Telemetry(pd.DataFrame):
"""Multi-channel time series telemetry data
The object can contain multiple telemetry channels. Multiple telemetry objects with different channels
can be merged on time. Each telemetry channel is one dataframe column.
Partial telemetry (e.g. for one lap only) can be obtained through various methods for slicing the data.
Additionally, methods for adding common computed data channels are available.
The following telemetry channels existed in the original API data:
- **Car data**:
- `Speed` (float): Car speed
- `RPM` (int): Car RPM
- `nGear` (int): Car gear number
- `Throttle` (float): 0-100 Throttle pedal pressure
- `Brake` (float): 0-100 Brake pedal pressure
- `DRS` (int): DRS indicator (See :meth:`car_data` for more info)
- **Position data**:
- `X` (float): X position
- `Y` (float): Y position
- `Z` (float): Z position
- `Status` (string): Flag - OffTrack/OnTrack
- **For both of the above**:
- `Time` (timedelta): Time (0 is start of the data slice)
- `SessionTime` (timedelta): Time elapsed since the start of the session
- `Date` (datetime): The full date + time at which this sample was created
- `Source` (str): Flag indicating how this sample was created:
- 'car': sample from original api car data
- 'pos': sample from original api position data
- 'interpolated': this sample was artificially created; all values are computed/interpolated
Example:
A sample's source is indicated as 'car'. It contains
values for speed, rpm and x, y, z coordinates.
Originally, this sample (with its timestamp) was received
when loading car data.
This means that the speed and rpm value are original
values as received from the api. The coordinates are
interpolated for this sample.
All methods of :class:`Telemetry` which resample or
interpolate data will preserve and adjust the source flag
correctly when modifying data.
Through merging/slicing it is possible to obtain any combination of telemetry channels!
The following additional computed data channels can be added:
- Distance driven between two samples:
:meth:`add_differential_distance`
- Distance driven since the first sample:
:meth:`add_distance`
- Relative distance driven since the first sample:
:meth:`add_relative_distance`
- Distance to driver ahead and car number of said driver:
:meth:`add_driver_ahead`
.. note:: See the separate explanation concerning the various definitions of 'Time' for more information on the
three date and time related channels: :ref:`time-explanation`
Slicing this class will return :class:`Telemetry` again for slices containing multiple rows. Single rows will be
returned as :class:`pandas.Series`.
Args:
*args (any): passed through to `pandas.DataFrame` superclass
session (:class:`Session`): Instance of associated session object. Required for full functionality!
driver (str): Driver number as string. Required for full functionality!
**kwargs (any): passed through to `pandas.DataFrame` superclass
"""
TELEMETRY_FREQUENCY = 'original'
"""Defines the frequency used when resampling the telemetry data. Either
the string ``'original'`` or an integer to specify a frequency in Hz."""
_CHANNELS = {
'X': {'type': 'continuous', 'missing': 'quadratic'},
'Y': {'type': 'continuous', 'missing': 'quadratic'},
'Z': {'type': 'continuous', 'missing': 'quadratic'},
'Status': {'type': 'discrete'},
'Speed': {'type': 'continuous', 'missing': 'linear'}, # linear is often required as quadratic overshoots
'RPM': {'type': 'continuous', 'missing': 'linear'}, # on sudden changes like sudden pedal application)
'Throttle': {'type': 'continuous', 'missing': 'linear'},
'Brake': {'type': 'discrete'},
'DRS': {'type': 'discrete'},
'nGear': {'type': 'discrete'},
'Source': {'type': 'excluded'}, # special case, custom handling
'Date': {'type': 'excluded'}, # special case, used as the index during resampling
'Time': {'type': 'excluded'}, # special case, Time/SessionTime recalculated from 'Date'
'SessionTime': {'type': 'excluded'},
'Distance': {'type': 'continuous', 'missing': 'quadratic'},
'RelativeDistance': {'type': 'continuous', 'missing': 'quadratic'},
'DifferentialDistance': {'type': 'continuous', 'missing': 'quadratic'},
'DriverAhead': {'type': 'discrete'},
'DistanceToDriverAhead': {'type': 'continuous', 'missing': 'linear'}
}
"""Known telemetry channels which are supported by default"""
_metadata = ['session', 'driver']
def __init__(self, *args, session=None, driver=None, **kwargs):
super().__init__(*args, **kwargs)
self.session = session
self.driver = driver
@property
def _constructor(self):
def _new(*args, **kwargs):
return Telemetry(*args, **kwargs).__finalize__(self)
return _new
@property
def base_class_view(self):
"""For a nicer debugging experience; can view DataFrame through this property in various IDEs"""
return pd.DataFrame(self)
[docs] def join(self, *args, **kwargs):
"""Wraps :mod:`pandas.DataFrame.join` and adds metadata propagation.
When calling `self.join` metadata will be propagated from self to the joined dataframe.
"""
meta = dict()
for var in self._metadata:
meta[var] = getattr(self, var)
ret = super().join(*args, **kwargs)
for var, val in meta.items():
setattr(ret, var, val)
return ret
[docs] def merge(self, *args, **kwargs):
"""Wraps :mod:`pandas.DataFrame.merge` and adds metadata propagation.
When calling `self.merge` metadata will be propagated from self to the merged dataframe.
"""
meta = dict()
for var in self._metadata:
meta[var] = getattr(self, var)
ret = super().merge(*args, **kwargs)
for var, val in meta.items():
setattr(ret, var, val)
return ret
[docs] def slice_by_mask(self, mask, pad=0, pad_side='both'):
"""Slice self using a boolean array as a mask.
Args:
mask (array-like): Array of boolean values with the same length as self
pad (int): Number of samples used for padding the sliced data
pad_side (str): Where to pad the data; possible options: 'both', 'before', 'after'
Returns:
:class:`Telemetry`
"""
if pad:
if pad_side in ('both', 'before'):
i_left_pad = max(0, np.min(np.where(mask)) - pad)
else:
i_left_pad = np.min(np.where(mask))
if pad_side in ('both', 'after'):
i_right_pad = min(len(mask), np.max(np.where(mask)) + pad)
else:
i_right_pad = np.max(np.where(mask))
mask[i_left_pad: i_right_pad + 1] = True
data_slice = self.loc[mask].copy()
return data_slice
[docs] def slice_by_lap(self, ref_laps, pad=0, pad_side='both', interpolate_edges=False):
"""Slice self to only include data from the provided lap or laps.
.. note:: Self needs to contain a 'SessionTime' column.
.. note:: When slicing with an instance of :class:`Laps` as a reference, the data will be sliced by first and
last lap. Missing laps in between will not be considered and data for these will still be included in
the sliced result.
Args:
ref_laps (Lap or Laps): The lap/laps by which to slice self
pad (int): Number of samples used for padding the sliced data
pad_side (str): Where to pad the data; possible options: 'both', 'before', 'after
interpolate_edges (bool): Add an interpolated sample at the beginning and end to exactly
match the provided time window.
Returns:
:class:`Telemetry`
"""
if isinstance(ref_laps, Laps) and len(ref_laps) > 1:
if 'DriverNumber' not in ref_laps.columns:
ValueError("Laps is missing 'DriverNumber'. Cannot return telemetry for unknown driver.")
if not len(ref_laps['DriverNumber'].unique()) <= 1:
raise ValueError("Cannot create telemetry for multiple drivers at once!")
end_time = ref_laps['Time'].max()
start_time = ref_laps['LapStartTime'].min()
elif isinstance(ref_laps, (Lap, Laps)):
if isinstance(ref_laps, Laps): # one lap in Laps
ref_laps = ref_laps.iloc[0] # needs to be handled as a single lap
if 'DriverNumber' not in ref_laps.index:
ValueError("Lap is missing 'DriverNumber'. Cannot return telemetry for unknown driver.")
end_time = ref_laps['Time']
start_time = ref_laps['LapStartTime']
else:
raise TypeError("Attribute 'ref_laps' needs to be an instance of `Lap` or `Laps`")
return self.slice_by_time(start_time, end_time, pad, pad_side, interpolate_edges)
[docs] def slice_by_time(self, start_time, end_time, pad=0, pad_side='both', interpolate_edges=False):
"""Slice self to only include data in a specific time frame.
.. note:: Self needs to contain a 'SessionTime' column. Slicing by time use the 'SessionTime' as its reference.
Args:
start_time (Timedelta): Start of the section
end_time (Timedelta): End of the section
pad (int): Number of samples used for padding the sliced data
pad_side (str): Where to pad the data; possible options: 'both', 'before', 'after
interpolate_edges (bool): Add an interpolated sample at the beginning and end to exactly
match the provided time window.
Returns:
:class:`Telemetry`
"""
if interpolate_edges:
edges = Telemetry({'SessionTime': (start_time, end_time),
'Date': (start_time + self.session.t0_date, end_time + self.session.t0_date)},
session=self.session)
d = self.merge_channels(edges)
else:
d = self.copy() # TODO no copy?
sel = ((d['SessionTime'] <= end_time) & (d['SessionTime'] >= start_time))
if np.any(sel):
data_slice = d.slice_by_mask(sel, pad, pad_side)
if 'Time' in data_slice.columns:
# shift time to 0 so laps can overlap
data_slice.loc[:, 'Time'] = data_slice['SessionTime'] - start_time
return data_slice
return Telemetry()
[docs] def merge_channels(self, other, frequency=None):
"""Merge telemetry objects containing different telemetry channels.
The two objects don't need to have a common time base. The data will be merged, optionally resampled and
missing values will be interpolated.
:attr:`Telemetry.TELEMETRY_FREQUENCY` determines if and how the data is resampled. This can be overridden using
the `frequency` keyword fo this method.
Merging and resampling:
If the frequency is 'original', data will not be resampled. The two objects will be merged and all
timestamps of both objects are kept. Values will be interpolated so that all telemetry channels contain
valid data for all timestamps. This is the default and recommended option.
If the frequency is specified as an integer in Hz the data will be merged as before. After that, the merged
time base will be resampled from the first value on at the specified frequency. Afterwards, the data will
be interpolated to fit the new time base. This means that usually most if not all values of the data will
be interpolated values. This is detrimental for overall accuracy.
Interpolation:
Missing values after merging will be interpolated for all known telemetry channels using
:meth:`fill_missing`. Different interpolation methods are used depending on what kind of data the channel
contains. For example, forward fill is used to interpolated 'nGear' while linear interpolation is used
for 'RPM' interpolation.
.. note :: Unknown telemetry channels will be merged but missing values will not be interpolated. This can
either be done manually or a custom telemetry channel can be added using :meth:`register_new_channel`.
.. note :: Do not resample data multiple times. Always resample based on the original data
to preserve accuracy
Args:
other (:class:`Telemetry` or :class:`pandas.DataFrame`): Object to be merged with self
frequency (str or int): Optional frequency to overwrite global preset. (Either string 'original' or integer
for a frequency in Hz)
Returns:
:class:`Telemetry`
"""
# merge the data and interpolate missing; 'Date' needs to be the index
data = self.set_index('Date')
other = other.set_index('Date')
# save dtypes before merging so they can be restored after merging
# necessary for example because merging produces NaN values which would cause an int column to become float
# but it can be converted back to int after interpolating missing values
dtype_map = dict()
for df in data, other:
for col in df.columns:
if col not in dtype_map.keys():
dtype_map[col] = df[col].dtype
# Exclude columns existing on both dataframes from one dataframe before merging (cannot merge with duplicates)
on_both_columns = set(other.columns).intersection(set(data.columns))
merged = other.merge(data[data.columns.difference(on_both_columns, sort=False)],
how='outer', left_index=True, right_index=True, sort=True)
# now use the previously excluded columns to update the missing values in the merged dataframe
for col in on_both_columns:
merged[col].update(data[col])
if 'Driver' in merged.columns and len(merged['Driver'].unique()) > 1:
raise ValueError("Cannot merge multiple drivers")
if not frequency:
frequency = data.TELEMETRY_FREQUENCY
i = data.get_first_non_zero_time_index()
if i is None:
raise ValueError("No valid 'Time' data. Cannot resample!")
ref_date = merged.index[i]
# data needs to be resampled/interpolated differently, depending on what kind of data it is
# how to handle which column is defined in self._CHANNELS
if frequency == 'original':
# no resampling but still interpolation due to merging
merged = merged.fill_missing()
merged = merged.reset_index().rename(columns={'index': 'Date'}) # make 'Date' a column again
else:
frq = f'{1 / frequency}S'
resampled_columns = dict()
for ch in self._CHANNELS.keys():
if ch not in merged.columns:
continue
sig_type = self._CHANNELS[ch]['type']
if sig_type == 'continuous':
missing = self._CHANNELS[ch]['missing']
res = merged.loc[:, ch] \
.resample(frq, origin=ref_date).mean().interpolate(method=missing, fill_value='extrapolate')
elif sig_type == 'discrete':
res = merged.loc[:, ch].resample(frq, origin=ref_date).ffill().ffill().bfill()
# first ffill is a method of the resampler object and will ONLY ffill values created during
# resampling but not already existing NaN values. NaN values already existed because of merging,
# therefore call ffill a second time as a method of the returned series to fill these too
# only use bfill after ffill to fix first row
else:
continue
resampled_columns[ch] = res
res_source = merged.loc[:, 'Source'].resample(frq, origin=ref_date).asfreq().fillna(value='interpolation')
resampled_columns['Source'] = res_source
# join resampled columns and make 'Date' a column again
merged = Telemetry(resampled_columns, session=self.session).reset_index().rename(columns={'index': 'Date'})
# recalculate the time columns
merged['SessionTime'] = merged['Date'] - self.session.t0_date
merged['Time'] = merged['SessionTime'] - merged['SessionTime'].iloc[0]
# restore data types from before merging
for col in dtype_map.keys():
try:
merged.loc[:, col] = merged.loc[:, col].astype(dtype_map[col])
except ValueError:
logging.warning(f"Failed to preserve data type for column '{col}' while merging telemetry.")
return merged
[docs] def resample_channels(self, rule=None, new_date_ref=None, **kwargs):
"""Resample telemetry data.
Convenience method for frequency conversion and resampling. Up and down sampling of data is supported.
'Date' and 'SessionTime' need to exist in the data. 'Date' is used as the main time reference.
There are two ways to use this method:
- Usage like :meth:`pandas.DataFrame.resample`: In this case you need to specify the 'rule' for resampling
and any additional keywords will be passed on to :meth:`pandas.Series.resample` to create a new time
reference. See the pandas method to see which options are available.
- using the 'new_date_ref' keyword a :class:`pandas.Series` containing new values for date
(dtype :class:`pandas.Timestamp`) can be provided. The existing data will be resampled onto this new
time reference.
Args:
rule (optional, str): Resampling rule for :meth:`pandas.Series.resample`
new_date_ref (optional, pandas.Series): New custom Series of reference dates
**kwargs (optional, any): Only in combination with 'rule'; additional parameters for
:meth:`pandas.Series.resample`
"""
if rule is not None and new_date_ref is not None:
raise ValueError("You can only specify one of 'rule' or 'new_index'")
if rule is None and new_date_ref is None:
raise ValueError("You need to specify either 'rule' or 'new_index'")
if new_date_ref is None:
st = pd.Series(index=pd.DatetimeIndex(self['Date']), dtype=int).resample(rule, **kwargs).asfreq()
new_date_ref = pd.Series(st.index)
new_tel = Telemetry(session=self.session, driver=self.driver, columns=self.columns)
new_tel.loc[:, 'Date'] = new_date_ref
combined_tel = self.merge_channels(Telemetry({'Date': new_date_ref}, session=self.session))
mask = combined_tel['Date'].isin(new_date_ref)
new_tel = combined_tel.loc[mask, :]
return new_tel
[docs] def fill_missing(self):
"""Calculate missing values in self.
Only known telemetry channels will be interpolated. Unknown channels are skipped and returned unmodified.
Interpolation will be done according to the default mapping and according to options specified for
registered custom channels. For example:
| Linear interpolation will be used for continuous values (Speed, RPM)
| Forward-fill will be used for discrete values (Gear, DRS, ...)
See :meth:`register_new_channel` for adding custom channels.
"""
ret = self.copy()
for ch in self._CHANNELS.keys():
if ch not in self.columns:
continue
sig_type = self._CHANNELS[ch]['type']
if sig_type == 'continuous': # yes, this is necessary to prevent pandas from crashing
if ret[ch].dtype == 'object':
warnings.warn("Interpolation not possible for telemetry "
"channel because dtype is 'object'")
missing = self._CHANNELS[ch]['missing']
ret.loc[:, ch] = ret.loc[:, ch] \
.interpolate(method=missing, limit_direction='both', fill_value='extrapolate')
elif sig_type == 'discrete':
ret.loc[:, ch] = ret.loc[:, ch].ffill().ffill().bfill()
# first ffill is a method of the resampler object and will ONLY ffill values created during
# resampling but not already existing NaN values. NaN values already existed because of merging,
# therefore call ffill a second time as a method of the returned series to fill these too
# only use bfill after ffill to fix first row
if 'Source' in ret.columns:
ret.loc[:, 'Source'] = ret.loc[:, 'Source'].fillna(value='interpolation')
if 'Date' in self.columns:
ret['SessionTime'] = ret['Date'] - self.session.t0_date
elif isinstance(ret.index, pd.DatetimeIndex):
ret['SessionTime'] = ret.index - self.session.t0_date # assume index is Date
ret['Time'] = ret['SessionTime'] - ret['SessionTime'].iloc[0]
return ret
[docs] @classmethod
def register_new_channel(cls, name, signal_type, interpolation_method=None):
"""Register a custom telemetry channel.
Registered telemetry channels are automatically interpolated when merging or resampling data.
Args:
name (str): Telemetry channel/column name
signal_type (str): One of three possible signal types:
- 'continuous': Speed, RPM, Distance, ...
- 'discrete': DRS, nGear, status values, ...
- 'excluded': Data channel will be ignored during resampling
interpolation_method (optional, str): The interpolation method
which should be used. Can only be specified and is required
in combination with ``signal_type='continuous'``. See
:meth:`pandas.Series.interpolate` for possible interpolation
methods.
"""
if signal_type not in ('discrete', 'continuous', 'excluded'):
raise ValueError(f"Unknown signal type {signal_type}.")
if signal_type == 'continuous' and interpolation_method is None:
raise ValueError("signal_type='continuous' requires interpolation_method to be specified.")
cls._CHANNELS[name] = {'type': signal_type, 'missing': interpolation_method}
[docs] def get_first_non_zero_time_index(self):
"""Return the first index at which the 'Time' value is not zero or NA/NaT"""
# find first row where time is not zero; usually this is the first row but sometimes.....
i_arr = np.where((self['Time'] != pd.Timedelta(0)) & ~pd.isna(self['Time']))[0]
if i_arr.size != 0:
return np.min(i_arr)
return None
[docs] def add_differential_distance(self, drop_existing=True):
"""Add column 'DifferentialDistance' to self.
This column contains the distance driven between subsequent samples.
Calls :meth:`calculate_differential_distance` and joins the result
with self.
Args:
drop_existing (bool): Drop and recalculate column if it already exists
Returns:
:class:`Telemetry`: self joined with new column or self if column exists and `drop_existing` is False.
"""
if ('DifferentialDistance' in self.columns) and not drop_existing:
return self
new_dif_dist = pd.DataFrame(
{'DifferentialDistance': self.calculate_differential_distance()}
)
if 'DifferentialDistance' in self.columns:
return self.drop(labels='DifferentialDistance', axis=1) \
.join(new_dif_dist, how='outer')
return self.join(new_dif_dist, how='outer')
[docs] def add_distance(self, drop_existing=True):
"""Add column 'Distance' to self.
This column contains the distance driven since the first sample of self in meters.
The data is produced by integrating the differential distance between subsequent laps.
You should not apply this function to telemetry of many laps simultaneously to reduce integration error.
Instead apply it only to single laps or few laps at a time!
Calls :meth:`integrate_distance` and joins the result with self.
Args:
drop_existing (bool): Drop and recalculate column if it already exists
Returns:
:class:`Telemetry`: self joined with new column or self if column exists and `drop_existing` is False.
"""
if ('Distance' in self.columns) and not drop_existing:
return self
new_dist = pd.DataFrame({'Distance': self.integrate_distance()})
if 'Distance' in self.columns:
return self.drop(labels='Distance', axis=1).join(new_dist, how='outer')
return self.join(new_dist, how='outer')
[docs] def add_relative_distance(self, drop_existing=True):
"""Add column 'RelativeDistance' to self.
This column contains the distance driven since the first sample as
a floating point number where ``0.0`` is the first sample of self
and ``1.0`` is the last sample.
This is calculated the same way as 'Distance' (see: :meth:`add_distance`). The same warnings apply.
Args:
drop_existing (bool): Drop and recalculate column if it already exists
Returns:
:class:`Telemetry`: self joined with new column or self if column exists and `drop_existing` is False.
"""
if 'RelativeDistance' in self.columns:
if drop_existing:
d = self.drop(labels='RelativeDistance', axis=1)
else:
return self
else:
d = self
if 'Distance' in d.columns:
rel_dist = d.loc[:, 'Distance'] / d.loc[:, 'Distance'].iloc[-1]
else:
dist = d.integrate_distance()
rel_dist = dist / dist.iloc[-1]
return d.join(pd.DataFrame({'RelativeDistance': rel_dist}), how='outer')
[docs] def add_driver_ahead(self, drop_existing=True):
"""Add column 'DriverAhead' and 'DistanceToDriverAhead' to self.
DriverAhead: Driver number of the driver ahead as string
DistanceToDriverAhead: Distance to next car ahead in meters
.. note:: Cars in the pit lane are currently not excluded from the data. They will show up when overtaken on
pit straight even if they're not technically in front of the car. A fix for this is TBD with other
improvements.
This should only be applied to data of single laps or few laps at a time to reduce integration error.
For longer time spans it should be applied per lap and the laps
should be merged afterwards.
If you absolutely need to apply it to a whole session, use the legacy implementation. Note that data of
the legacy implementation will be considerably less smooth. (see :mod:`fastf1.legacy`)
Calls :meth:`calculate_driver_ahead` and joins the result with self.
Args:
drop_existing (bool): Drop and recalculate column if it already exists
Returns:
:class:`Telemetry`: self joined with new column or self if column exists and `drop_existing` is False.
"""
if 'DriverAhead' in self.columns and 'DistanceToDriverAhead' in self.columns:
if drop_existing:
d = self.drop(labels='DriverAhead', axis=1) \
.drop(labels='DistanceToDriverAhead', axis=1)
else:
return self
else:
d = self
drv_ahead, dist = self.calculate_driver_ahead()
return d.join(pd.DataFrame({'DriverAhead': drv_ahead,
'DistanceToDriverAhead': dist},
index=d.index), how='outer')
[docs] def calculate_differential_distance(self):
"""Calculate the distance between subsequent samples of self.
Distance is in meters
Returns:
:class:`pandas.Series`
"""
if not all([col in self.columns for col in ('Speed', 'Time')]):
raise ValueError("Telemetry does not contain required channels 'Time' and 'Speed'.")
if self.size != 0:
dt = self['Time'].dt.total_seconds().diff()
dt.iloc[0] = self['Time'].iloc[0].total_seconds()
ds = self['Speed'] / 3.6 * dt
return ds
else:
return pd.Series()
[docs] def integrate_distance(self):
"""Return the distance driven since the first sample of self.
Distance is in meters. The data is produce by integration. Integration error will stack up when used for
long slices of data. This should therefore only be used for data of single laps or few laps at a time.
Returns:
:class:`pd.Series`
"""
ds = self.calculate_differential_distance()
if not ds.empty:
return ds.cumsum()
else:
return pd.Series()
[docs] def calculate_driver_ahead(self):
"""Calculate driver ahead and distance to driver ahead.
Driver ahead: Driver number of the driver ahead as string
Distance to driver ahead: Distance to the car ahead in meters
.. note:: This gives a smoother/cleaner result than the legacy implementation but WILL introduce
integration error when used over long distances (more than one or two laps may sometimes be considered
a long distance). If in doubt, do sanity checks (against the legacy version or in another way).
Returns:
driver ahead (numpy.array), distance to driver ahead (numpy.array)
"""
t_start = self['SessionTime'].iloc[0]
t_end = self['SessionTime'].iloc[-1]
combined_distance = pd.DataFrame()
# Assume the following lap profile as a catch all for all drivers
#
# |------ Lap before ------|------ n Laps between ------|------ Lap after ------|
# ^ ^
# t_start t_end
# Integration of the distance needs to start at the finish line so that there exists a common zero point
# Therefore find the "lap before" which is the lap during which the telemetry slice starts and the "lap after"
# where the telemetry slice ends
# Integrate distance over all relevant laps and slice by t_start and t_end after to get the interesting
# part only
own_laps = self.session.laps[self.session.laps['DriverNumber'] == self.driver]
first_lap_number = (own_laps[own_laps['LapStartTime'] <= t_start])['LapNumber'].iloc[-1]
for drv in self.session.drivers:
# find correct first relevant lap; very important for correct zero point in distance
drv_laps = self.session.laps[self.session.laps['DriverNumber'] == drv]
if drv_laps.empty: # Only include drivers who participated in this session
continue
drv_laps_before = drv_laps[(drv_laps['LapStartTime'] <= t_start)]
if not drv_laps_before.empty:
lap_n_before = drv_laps_before['LapNumber'].iloc[-1]
if lap_n_before < first_lap_number:
# driver is behind on track an therefore will cross the finish line AFTER self
# therefore above check for LapStartTime <= t_start is wrong
# the first relevant lap is the first lap with LapStartTime > t_start which is lap_n_before += 1
lap_n_before += 1
else:
lap_n_before = min(drv_laps['LapNumber'])
# find last relevant lap so as to no do too much unnecessary calculation later
drv_laps_after = drv_laps[drv_laps['Time'] >= t_end]
lap_n_after = drv_laps_after['LapNumber'].iloc[0] \
if not drv_laps_after.empty \
else max(drv_laps['LapNumber'])
relevant_laps = drv_laps[(drv_laps['LapNumber'] >= lap_n_before) & (drv_laps['LapNumber'] <= lap_n_after)]
if relevant_laps.empty:
continue
# first slice by lap and calculate distance, so that distance is zero at finish line
drv_tel = self.session.car_data[drv].slice_by_lap(relevant_laps).add_distance() \
.loc[:, ('SessionTime', 'Distance')].rename(columns={'Distance': drv})
# now slice again by time to only get the relevant time frame
drv_tel = drv_tel.slice_by_time(t_start, t_end)
if drv_tel.empty:
continue
drv_tel = drv_tel.set_index('SessionTime')
combined_distance = combined_distance.join(drv_tel, how='outer')
# create driver map for array
drv_map = combined_distance.loc[:, combined_distance.columns != self.driver].columns.to_numpy()
own_dst = combined_distance.loc[:, self.driver].to_numpy()
other_dst = combined_distance.loc[:, combined_distance.columns != self.driver].to_numpy()
# replace distance with nan if it does not change
# prepend first row before diff so that array size stays the same; but missing first sample because of that
other_dst[np.diff(other_dst, n=1, axis=0, prepend=other_dst[0, :].reshape((1, -1))) == 0] = np.nan
# resize own_dst to match shape of other_dst for easy subtraction
own_dst = np.repeat(own_dst.reshape((-1, 1)), other_dst.shape[1], axis=1)
delta_dst = other_dst - own_dst
delta_dst[np.isnan(delta_dst)] = np.inf # substitute nan with inf, else nan is returned as min
delta_dst[delta_dst < 0] = np.inf # remove cars behind so that neg numbers are not returned as min
index_ahead = np.argmin(delta_dst, axis=1)
drv_ahead = np.array([drv_map[i] for i in index_ahead])
drv_ahead[np.all(delta_dst == np.inf, axis=1)] = '' # remove driver from all inf rows
dist_to_drv_ahead = np.array([delta_dst[i, index_ahead[i]] for i in range(len(index_ahead))])
dist_to_drv_ahead[np.all(delta_dst == np.inf, axis=1)] = np.nan # remove value from all inf rows
return drv_ahead, dist_to_drv_ahead
[docs]class Weekend:
"""
.. deprecated:: 2.2
Use :class:`fastf1.events.Event` instead
"""
def __new__(cls, year, gp):
warnings.warn("`fastf1.core.Weekend` has been deprecated and will be"
"removed in a future version.\n"
"Use `fastf1.events.Event` instead.", FutureWarning)
from fastf1 import events
return events.get_event(year, gp)
[docs]class Session:
"""Object for accessing session specific data.
The session class will usually be your starting point. This object will
have various information about the session.
.. note:: Most of the data is only available after calling
:func:`Session.load`
"""
def __init__(self, event, session_name, f1_api_support=False):
# TODO: load drivers immediately
# TODO: load driver list for older seasons through ergast
self.event = event
""":class:`~fastf1.events.Event`: Reference to the associated event
object."""
self.name = session_name
"""str: Name of this session, for example 'Qualifying', 'Race', 'FP1', ..."""
self.f1_api_support = f1_api_support
"""bool: The official F1 API supports this event and lap timing data and
telemetry data are available."""
self.date = self.event.get_session_date(session_name)
"""pandas.Datetime: Date at which this session took place."""
self.api_path = api.make_path(
self.event['EventName'],
self.event['EventDate'].strftime('%Y-%m-%d'),
self.name, self.date.strftime('%Y-%m-%d')
)
"""str: API base path for this session"""
self._session_status = dict()
self._laps: Laps
self._t0_date: pd.Timestamp
self._session_start_time: pd.Timedelta
self._car_data = dict()
self._pos_data = dict()
self._weather_data: pd.DataFrame
self._results: SessionResults
def _get_property_warn_not_loaded(self, name):
d = getattr(self, name, None)
if d is None:
raise DataNotLoadedError("The data you are trying to access has not "
"been loaded yet. See `Session.load`")
return d
@property
def weekend(self):
"""Deprecated: use :attr:`Session.event` instead"""
warnings.warn("The property `Session.weekend` has been renamed to "
"`Session.event`.\n The old property will be removed in"
"a future version.", FutureWarning)
return self.event
@property
def drivers(self):
""":class:`list`: List of all drivers that took part in this
session; contains driver numbers as string.
Data is available after calling `Session.load`
"""
return list(self.results['DriverNumber'])
@property
def results(self):
""":class:`SessionResults`: Session result with driver information.
Data is available after calling `Session.load`
"""
return self._get_property_warn_not_loaded('_results')
@property
def laps(self):
""":class:`Laps`: All laps from all drivers driven in this session.
Data is available after calling `Session.load` with ``laps=True``
"""
return self._get_property_warn_not_loaded('_laps')
@property
def weather_data(self):
"""Dataframe containing weather data for this session as received
from the api. See :func:`fastf1.api.weather_data` for available data
channels. Each data channel is one row of the dataframe.
Data is available after calling `Session.load` with ``weather=True``
"""
return self._get_property_warn_not_loaded('_weather_data')
@property
def car_data(self):
"""Dictionary of car telemetry (Speed, RPM, etc.) as received from
the api by car number (where car number is a string and the telemetry
is an instance of :class:`Telemetry`)
Data is available after calling `Session.load` with ``telemetry=True``
"""
return self._get_property_warn_not_loaded('_car_data')
@property
def pos_data(self):
"""Dictionary of car position data as received from the api by car
number (where car number is a string and the telemetry
is an instance of :class:`Telemetry`)
Data is available after calling `Session.load` with ``telemetry=True``
"""
return self._get_property_warn_not_loaded('_pos_data')
@property
def session_status(self):
""":class:`pandas.Dataframe`: Session status data as returned by
:func:`fastf1.api.session_status_data`
Data is available after calling `Session.load` with ``laps=True``
"""
return self._get_property_warn_not_loaded('_session_status')
@property
def session_start_time(self):
""":class:`pandas.Timedelta`: Session time at which the session was
started according to the session status data. This is not the
time at which the session is scheduled to be started!
Data is available after calling `Session.load` with ``laps=True``
"""
return self._get_property_warn_not_loaded('_session_start_time')
@property
def t0_date(self):
""":class:`pandas.Datetime`: Date timestamp which marks the beginning
of the data stream (the moment at which the session time is zero).
Data is available after calling `Session.load` with ``telemetry=True``
"""
return self._get_property_warn_not_loaded('_t0_date')
[docs] def load(self, *, laps=True, telemetry=True, weather=True,
livedata=None):
"""Load session data from the supported APIs.
This method allows to flexibly load some or all data that FastF1 can
give you access to. Without specifying any further options, all data
is loaded by default.
Downloading and parsing of the data takes a considerable amount of
time. Therefore, it is highly recommended to enable caching so that
most of the data processing needs to be done only once.
.. note:: **Lap data: drivers crashing and retiring**
*During a session:*
An additional last lap is added for a driver if the last timed
lap of a driver is not an inlap and the session is aborted next.
The `Time` for when this lap was "set" will be set to the time at
which the session was aborted.
*First lap in a race:*
A single lap with minimal information will be added in race
sessions if a driver does not complete at least one timed lap.
The `LapStartTime` for this lap will be set to the start time
of the session as with all other laps in a race. The `Time` at
which this lap was "set" will be set to the time at which the
first driver completes their first lap.
.. note:: Absolute time is not super accurate. The moment a lap
is logged is not always the same and there will be some
jitter. At the moment lap time reference is synchronised
on the sector time triggered with lowest latency.
Expect an error of around ±10m when overlapping telemetry
data of different laps.
Args:
laps (bool): Load laps and session status data.
telemetry (bool): Load telemetry data.
weather (bool): Load weather data.
livedata (:class:`fastf1.livetiming.data.LiveTimingData`, optional) :
instead of requesting the data from the api, locally saved
livetiming data can be used as a data source
"""
logging.info(f"Loading data for "
f"{self.event['EventName']} - {self.name}"
f" [v{fastf1.__version__}]")
self._load_drivers_results(livedata=livedata)
if self.f1_api_support:
if laps:
try:
self._load_laps_data(livedata)
except Exception as exc:
logging.warning("Failed to load lap data!")
logging.debug("Lap data failure traceback:", exc_info=exc)
if telemetry:
try:
self._load_telemetry(livedata=livedata)
except Exception as exc:
logging.warning("Failed to load telemetry data!")
logging.debug("Telemetry data failure traceback:", exc_info=exc)
if weather:
try:
self._load_weather_data(livedata=livedata)
except Exception as exc:
logging.warning("Failed to load weather data!")
logging.debug("Weather data failure traceback:", exc_info=exc)
else:
if any((laps, telemetry, weather)):
logging.warning(
"Cannot load laps, telemetry and weather data because "
"the relevant API is not supported for this session."
)
logging.info(f"Finished loading data for {len(self.drivers)} "
f"drivers: {self.drivers}")
[docs] def load_laps(self, with_telemetry=False, livedata=None):
"""
.. deprecated:: 2.2
use :func:`Session.load` instead
"""
# TODO: remove in v2.3
warnings.warn("`Session.load_laps` is deprecated and will be"
"removed in a future version.\n"
"Use `Session.load` instead.", FutureWarning)
self.load(telemetry=with_telemetry, livedata=livedata)
return self.laps
[docs] def load_telemetry(self, livedata=None):
"""
.. deprecated:: 2.2
use :func:`Session.load` instead
"""
# TODO: remove in v2.3
warnings.warn("`Session.load_laps` is deprecated and will be"
"removed in a future version.\n"
"Use `Session.load` instead.", FutureWarning)
self._load_telemetry(livedata=livedata)
def _load_laps_data(self, livedata):
data, _ = api.timing_data(self.api_path, livedata=livedata)
app_data = api.timing_app_data(self.api_path, livedata=livedata)
logging.info("Processing timing data...")
# Matching data and app_data. Not super straightforward
# Sometimes a car may enter the pit without changing tyres, so
# new compound is associated with the help of logging time.
useful = app_data[['Driver', 'Time', 'Compound', 'TotalLaps', 'New']]
useful = useful[~useful['Compound'].isnull()]
# check when a session was started; for a race this indicates the
# start of the race
session_status = api.session_status_data(self.api_path,
livedata=livedata)
for i in range(len(session_status['Status'])):
if session_status['Status'][i] == 'Started':
self._session_start_time = session_status['Time'][i]
break
self._session_status = pd.DataFrame(session_status)
df = None
drivers = self.drivers
if not drivers:
# no driver list, generate from lap data
drivers = set(data['Driver'].unique())\
.intersection(set(useful['Driver'].unique()))
_nums_df = pd.DataFrame({'DriverNumber': list(drivers)},
index=list(drivers))
_info_df = pd.DataFrame(fastf1._DRIVER_TEAM_MAPPING).T
self._results = SessionResults(_nums_df.join(_info_df),
force_default_cols=True)
logging.warning("Generating minimal driver "
"list from timing data.")
for i, driver in enumerate(drivers):
d1 = data[data['Driver'] == driver]
d2 = useful[useful['Driver'] == driver]
only_one_lap = False
if not len(d1):
if (self.name == 'Race') and len(d2):
# add data for drivers who crashed on the very first lap
# as a downside, this potentially adds a nonexistent lap
# for drivers who could not start the race
only_one_lap = True
result = d1.copy()
result['Driver'] = [driver, ]
result['NumberOfLaps'] = 0
result['NumberOfPitStops'] = 0
result['Time'] = data['Time'].min()
result['IsPersonalBest'] = False
result['Compound'] = d2['Compound'].iloc[0]
result['TotalLaps'] = d2['TotalLaps'].iloc[0]
result['New'] = d2['New'].iloc[0]
else:
logging.warning(f"No lap data for driver {driver}")
continue # no data for this driver; skip
elif not len(d2):
result = d1.copy()
result['Compound'] = str()
result['TotalLaps'] = np.nan
result['New'] = False
logging.warning(f"No tyre data for driver {driver}")
else:
result = pd.merge_asof(d1, d2, on='Time', by='Driver')
# calculate lap start time by setting it to the 'Time' of the previous lap
laps_start_time = list(result['Time'])[:-1]
if self.name == 'Race':
# assumption that the first lap started when the session was started can only be made for the race
laps_start_time.insert(0, self.session_start_time)
else:
laps_start_time.insert(0, pd.NaT)
result.loc[:, 'LapStartTime'] = pd.Series(
laps_start_time, dtype='timedelta64[ns]'
)
# set missing lap start times to pit out time where possible
mask = pd.isna(result['LapStartTime']) & (~pd.isna(result['PitOutTime']))
result.loc[mask, 'LapStartTime'] = result.loc[mask, 'PitOutTime']
# create total laps counter
for npit in result['NumberOfPitStops'].unique():
sel = result['NumberOfPitStops'] == npit
result.loc[sel, 'TotalLaps'] += np.arange(0, sel.sum()) + 1
# check if there is another lap during which the session was aborted
# but which is not in the data
# if yes, add as much data as possible for it
# set the time of abort as lap end time given that there is no
# accurate time available
# this block of code has no tests; testing would require to mock
# the data as the actual data may be updated on the server after
# some time and the problem no longer occurs
if pd.isna(result['PitInTime'].iloc[-1]) and not only_one_lap:
if not pd.isna(result['Time'].iloc[-1]):
next_statuses = self.session_status[
self.session_status['Time'] > result['Time'].iloc[-1]
]
else:
next_statuses = self.session_status[
self.session_status['Time']
> result['LapStartTime'].iloc[-1]
]
aborted = False
if not next_statuses.empty:
next_status = next_statuses.iloc[0]
aborted = (next_status['Status'] == 'Aborted')
if aborted:
new_last = pd.DataFrame({
'LapStartTime': [result['Time'].iloc[-1]],
'Time': [next_status['Time']],
'Driver': [result['Driver'].iloc[-1]],
'NumberOfLaps': [result['NumberOfLaps'].iloc[-1] + 1],
'NumberOfPitStops':
[result['NumberOfPitStops'].iloc[-1]],
# 'IsPersonalBest': False,
'Compound': [result['Compound'].iloc[-1]],
'TotalLaps': [result['TotalLaps'].iloc[-1] + 1],
'New': [result['New'].iloc[-1]],
})
if not only_one_lap:
result = result.append(new_last).reset_index(drop=True)
else:
result = new_last
df = pd.concat([df, result], sort=False)
if df is None:
raise NoLapDataError
laps = df.reset_index(drop=True) # noqa: F821
laps.rename(columns={'TotalLaps': 'TyreLife',
'NumberOfPitStops': 'Stint',
'Driver': 'DriverNumber',
'NumberOfLaps': 'LapNumber',
'New': 'FreshTyre'}, inplace=True)
laps['Stint'] += 1 # counting stints from 1
t_map = {r['DriverNumber']: r['TeamName']
for _, r in self.results.iterrows()}
laps['Team'] = laps['DriverNumber'].map(t_map)
d_map = {r['DriverNumber']: r['Abbreviation']
for _, r in self.results.iterrows()}
laps['Driver'] = laps['DriverNumber'].map(d_map)
# add track status data
ts_data = api.track_status_data(self.api_path, livedata=livedata)
laps['TrackStatus'] = '1'
def applicator(new_status, current_status):
if current_status == '1':
return new_status
elif new_status not in current_status:
return current_status + new_status
else:
return current_status
if len(ts_data['Time']) > 0:
t = ts_data['Time'][0]
status = ts_data['Status'][0]
for next_t, next_status in zip(ts_data['Time'][1:], ts_data['Status'][1:]):
if status != '1':
# status change partially in lap partially outside
sel = (((next_t >= laps['LapStartTime']) & (laps['LapStartTime'] >= t)) |
((t <= laps['Time']) & (laps['Time'] <= next_t)))
laps.loc[sel, 'TrackStatus'] = laps.loc[sel, 'TrackStatus'].apply(
lambda curr: applicator(status, curr)
)
# status change two times in one lap (e.g. short yellow flag)
sel = ((laps['LapStartTime'] <= t) & (laps['Time'] >= next_t))
laps.loc[sel, 'TrackStatus'] = laps.loc[sel, 'TrackStatus'].apply(
lambda curr: applicator(status, curr)
)
t = next_t
status = next_status
sel = laps['LapStartTime'] >= t
laps.loc[sel, 'TrackStatus'] = laps.loc[sel, 'TrackStatus'].apply(
lambda curr: applicator(status, curr)
)
else:
logging.warning("Could not load any valid session status information!")
self._laps = Laps(laps, session=self)
self._check_lap_accuracy()
def _check_lap_accuracy(self):
"""Accuracy validation; simples yes/no validation
Currently only relies on provided information which can't catch all problems"""
# TODO: check for outliers in lap start position
# self.laps['IsAccurate'] = False # default should be not accurate
for drv in self.drivers:
is_accurate = list()
prev_lap = None
integrity_errors = 0
for _, lap in self.laps[self.laps['DriverNumber'] == drv].iterrows():
# require existence, non-existence and specific values for some variables
check_1 = (pd.isnull(lap['PitInTime'])
& pd.isnull(lap['PitOutTime'])
& (lap['TrackStatus'] in ('1', '2')) # slightly paranoid, allow only green and yellow flag
& (not pd.isnull(lap['LapTime']))
& (not pd.isnull(lap['Sector1Time']))
& (not pd.isnull(lap['Sector2Time']))
& (not pd.isnull(lap['Sector3Time'])))
if check_1: # only do check 2 if all necessary values for this check are even available
# sum of sector times should be almost equal to lap time (tolerance 3ms)
check_2 = np.allclose(np.sum((lap['Sector1Time'], lap['Sector2Time'],
lap['Sector3Time'])).total_seconds(),
lap['LapTime'].total_seconds(),
atol=0.003, rtol=0, equal_nan=False)
if not check_2:
integrity_errors += 1
else:
check_2 = False # data not available means fail
if prev_lap is not None:
# first lap after safety car often has timing issues (as do all laps under safety car)
check_3 = (prev_lap['TrackStatus'] != '4')
else:
check_3 = True # no previous lap, no SC error
result = check_1 and check_2 and check_3
is_accurate.append(result)
prev_lap = lap
if len(is_accurate) > 0:
self._laps.loc[self.laps['DriverNumber'] == drv, 'IsAccurate'] = is_accurate
if integrity_errors > 0:
logging.warning(f"Driver {drv: >2}: Lap timing integrity check failed for {integrity_errors} lap(s)")
def _load_drivers_results(self, *, livedata=None):
# get list of drivers
driver_info = None
if self.f1_api_support:
# load driver info from f1 api
driver_info = self._drivers_from_f1_api(livedata=livedata)
if not driver_info:
if not self.event.is_testing():
# load driver info and results from ergast
# (season 2017 and older or fallback from f1 api)
driver_info = self._drivers_results_from_ergast(
load_drivers=True, load_results=True
)
self._results = SessionResults(
driver_info, index=driver_info['DriverNumber'],
force_default_cols=True
)
else:
logging.warning("Failed to load driver list and "
"session results!")
self._results = SessionResults(force_default_cols=True)
else:
# extend existing driver info (f1 api) with results from ergast
drivers = pd.DataFrame(driver_info,
index=driver_info['DriverNumber'])
if not self.event.is_testing():
r = self._drivers_results_from_ergast(load_results=True)
else:
r = None
if r:
# join driver info and session results
results = pd.DataFrame(r).set_index('DriverNumber')
self._results = SessionResults(drivers.join(results),
force_default_cols=True)
else:
# return driver info without session results
self._results = SessionResults(drivers,
force_default_cols=True)
if 'Position' in self._results:
self._results = self._results.sort_values('Position')
def _drivers_from_f1_api(self, *, livedata=None):
try:
f1di = api.driver_info(self.api_path, livedata=livedata)
except Exception as exc:
logging.warning("Failed to load extended driver information!")
logging.debug("Exception while loading driver list", exc_info=exc)
driver_info = {}
else:
driver_info = collections.defaultdict(list)
for key1, key2 in {
'RacingNumber': 'DriverNumber',
'BroadcastName': 'BroadcastName',
'Tla': 'Abbreviation', 'TeamName': 'TeamName',
'TeamColour': 'TeamColor', 'FirstName': 'FirstName',
'LastName': 'LastName'
}.items():
for entry in f1di.values():
driver_info[key2].append(entry.get(key1))
if 'FirstName' in driver_info and 'LastName' in driver_info:
for first, last in zip(driver_info['FirstName'],
driver_info['LastName']):
driver_info['FullName'].append(f"{first} {last}")
return driver_info
def _drivers_results_from_ergast(self, *, load_drivers=False,
load_results=False):
if self.name in ('Qualifying', 'Sprint Qualifying', 'Race'):
session_name = self.name
else:
# this is a practice session, use drivers from race session but
# don't load results
session_name = 'Race'
load_results = False
d = collections.defaultdict(list)
try:
data = ergast.fetch_results(
self.event.year, self.event['RoundNumber'], session_name
)
except Exception as exc:
logging.warning("Failed to load data from Ergast API! "
"(This is expected for recent sessions)")
logging.debug("Ergast failure traceback:", exc_info=exc)
return d
time0 = None
for r in data:
d['DriverNumber'].append(r.get('number'))
if load_drivers:
d['Abbreviation'].append(
recursive_dict_get(r, 'Driver', 'code',
default_none=True))
first_name = recursive_dict_get(r, 'Driver', 'givenName',
default_none=True)
last_name = recursive_dict_get(r, 'Driver', 'familyName',
default_none=True)
d['FirstName'].append(first_name)
d['LastName'].append(last_name)
d['FullName'].append(f"{first_name} {last_name}")
d['TeamName'].append(
recursive_dict_get(r, 'Constructor', 'name',
default_none=True))
if load_results:
d['Position'].append(r.get('position'))
d['GridPosition'].append(r.get('grid'))
d['Q1'].append(to_timedelta(r.get('Q1')))
d['Q2'].append(to_timedelta(r.get('Q2')))
d['Q3'].append(to_timedelta(r.get('Q3')))
if time0 is None:
ts = recursive_dict_get(r, 'Time', 'time',
default_none=True)
if ts:
time0 = to_timedelta(ts)
else:
time0 = pd.NaT
d['Time'].append(time0)
else:
ts = recursive_dict_get(r, 'Time', 'time',
default_none=True)
if ts:
dt = to_timedelta(ts)
else:
dt = pd.NaT
d['Time'].append(time0 + dt)
d['Status'].append(r.get('status'))
d['Points'].append(r.get('points'))
return d
def _load_weather_data(self, livedata=None):
weather_data = api.weather_data(self.api_path, livedata=livedata)
weather_df = pd.DataFrame(weather_data)
self._weather_data = weather_df
def _load_telemetry(self, livedata=None):
"""Load telemetry data from the API.
This method can only be called after :meth:`load_laps` has been
called. You will usually just want to call :meth:`load_laps` with
the optional ``with_telemetry=True`` argument instead of calling this
method separately. The result will be the same.
The raw data is divided into car data (Speed, RPM, ...) and position data (coordinates, on/off track). For each
of the two types an instance of :class:`Telemetry` is created per driver. The properties
:attr:`Session.car_data` and :attr:`Session.pos_data` are dictionaries which hold the the `Telemetry` objects
keyed by driver number.
The telemetry data can either be accessed through the above mentioned attributes or conveniently on a per
lap basis through :class:`Lap` and :class:`Laps`. See :class:`Telemetry` on how to work with the telemetry
data.
Note that this method additionally calculates :attr:`Session.t0_date` and adds a `LapStartDate` column to
:attr:`Session.laps`.
Args:
livedata (:class:`fastf1.livetiming.data.LiveTimingData`, optional) :
instead of requesting the data from the api, locally saved
livetiming data can be used as a data source
"""
car_data = api.car_data(self.api_path, livedata=livedata)
pos_data = api.position_data(self.api_path, livedata=livedata)
self._calculate_t0_date(car_data, pos_data)
for drv in self.drivers:
try:
# drop and recalculate time stamps based on 'Date', because 'Date' has a higher resolution
drv_car = Telemetry(car_data[drv].drop(labels='Time', axis=1),
session=self, driver=drv)
drv_pos = Telemetry(pos_data[drv].drop(labels='Time', axis=1),
session=self, driver=drv)
except KeyError:
# not pos data or car data exists for this driver
continue
drv_car['Date'] = drv_car['Date'].round('ms')
drv_pos['Date'] = drv_pos['Date'].round('ms')
drv_car['Time'] = drv_car['Date'] - self.t0_date # create proper continuous timestamps
drv_pos['Time'] = drv_pos['Date'] - self.t0_date
drv_car['SessionTime'] = drv_car['Time']
drv_pos['SessionTime'] = drv_pos['Time']
self._car_data[drv] = drv_car
self._pos_data[drv] = drv_pos
self._laps['LapStartDate'] = self._laps['LapStartTime'] + self.t0_date
[docs] def get_driver(self, identifier):
"""
Get a driver object which contains additional information about a driver.
Args:
identifier (str): driver's three letter identifier (for
example 'VER') or driver number as string
Returns:
instance of :class:`Driver`
"""
mask = ((self.results['Abbreviation'] == identifier)
| (self.results['DriverNumber'] == identifier))
if not mask.any():
raise ValueError(f"Invalid driver identifier '{identifier}'")
return self.results[mask].iloc[0]
def _calculate_t0_date(self, car_data, pos_data):
"""Calculate the date timestamp at which data for this session is starting.
This does not mark the start of a race (or other sessions). This marks the start of the data which is sometimes
far before.
This function sets :attr:`self.t0_date` which is an internally required offset for some calculations.
The current assumption is that the latest date which can be calculated is correct. (Based on the timestamp with
the least delay.)
Args:
car_data: Car telemetry; should contain all samples and only original ones
pos_data: Car position data; should contain all samples and only original ones
"""
date_offset = None
for data in (car_data, pos_data):
for drv in data.keys():
new_offset = max(data[drv]['Date'] - data[drv]['Time'])
if date_offset is None or new_offset > date_offset:
date_offset = new_offset
self._t0_date = date_offset.round('ms')
[docs]class Laps(pd.DataFrame):
"""Object for accessing lap (timing) data of multiple laps.
Args:
*args (any): passed through to :class:`pandas.DataFrame` super class
session (:class:`Session`): instance of session class; required for
full functionality
**kwargs (any): passed through to :class:`pandas.DataFrame`
super class
This class allows for easily picking specific laps from all laps in a session. It implements some additional
functionality on top off the usual `pandas.DataFrame` functionality. Among others, the laps' associated telemetry
data can be accessed.
If for example you want to get the fastest lap of Bottas you can narrow it down like this::
import fastf1
session = fastf1.get_session(2019, 'Bahrain', 'Q')
session.load()
best_bottas = session.laps.pick_driver('BOT').pick_fastest()
print(best_bottas['LapTime'])
# Timedelta('0 days 00:01:28.256000')
Slicing this class will return :class:`Laps` again for slices containing multiple rows. Single rows will be
returned as :class:`Lap`.
The following information is available per lap (one DataFrame column for each):
- **Time** (pandas.Timedelta): Session time when the lap time was set (end of lap)
- **Driver** (string): Three letter driver identifier
- **DriverNumber** (str): Driver number
- **LapTime** (pandas.Timedelta): Recorded lap time.
Officially deleted lap times will *not* be deleted here.
Deleting laps is currently not supported.
- **LapNumber** (int): Recorded lap number
- **Stint** (int): Stint number
- **PitOutTime** (pandas.Timedelta): Session time when car exited the pit
- **PitInTime** (pandas.Timedelta): Session time when car entered the pit
- **Sector1Time** (pandas.Timedelta): Sector 1 recorded time
- **Sector2Time** (pandas.Timedelta): Sector 2 recorded time
- **Sector3Time** (pandas.Timedelta): Sector 3 recorded time
- **Sector1SessionTime** (pandas.Timedelta): Session time when the Sector 1 time was set
- **Sector2SessionTime** (pandas.Timedelta): Session time when the Sector 2 time was set
- **Sector3SessionTime** (pandas.Timedelta): Session time when the Sector 3 time was set
- **SpeedI1** (float): Speedtrap sector 1
- **SpeedI2** (float): Speedtrap sector 2
- **SpeedFL** (float): Speedtrap at finish line
- **SpeedST** (float): Speedtrap on longest straight (Not sure)
- **IsPersonalBest** (bool): Flag that indicates whether this lap is
the official personal best lap of a driver. If any lap of a driver
is quicker than their respective personal best lap, this means that
the quicker lap is invalid and not counted. This can happen it the
track limits were execeeded, for example.
- **Compound** (str): Tyre compound name: SOFT, MEDIUM ..
- **TyreLife** (float): Laps driven on this tire (includes laps in other sessions for used sets of tires)
- **FreshTyre** (bool): Tyre had TyreLife=0 at stint start, i.e. was a new tire
- **Team** (str): Team name
- **LapStartTime** (pandas.Timedelta): Session time at the start of the lap
- **LapStartDate** (pandas.Timestamp): Timestamp at the start of the lap
- **TrackStatus** (str): A string that contains track status numbers for all track status that occurred
during this lap. The meaning of the track status numbers is explained in
:func:`fastf1.api.track_status_data`. (Currently, track status data is only implemented per lap.
If a finer resolution is desired, you need to directly use the data returned by
:func:`fastf1.api.track_status_data`) # TODO updated when implemented
- **IsAccurate** (bool): If True, the lap has passed a basic accuracy check for timing data.
This does not guarantee accuracy but laps marked as inaccurate need to be handled with
caution. They might contain errors which can not be spotted easily.
Laps need to satisfy the following criteria to be marked
as accurate:
- not an inlap or outlap
- set under green or yellow flag (the api sometimes has issues
with data from SC/VSC laps)
- is not the first lap after a safety car period
(issues with SC/VSC might still apear on the first lap
after it has ended)
- has a value for lap time and all sector times
- the sum of the sector times matches the lap time
(will be additionally logged as integrity error if not)
"""
_metadata = ['session']
QUICKLAP_THRESHOLD = 1.07
"""Used to determine 'quick' laps. Defaults to the 107% rule."""
def __init__(self, *args, session=None, **kwargs):
super().__init__(*args, **kwargs)
self.session = session
@property
def _constructor(self):
def _new(*args, **kwargs):
return Laps(*args, **kwargs).__finalize__(self)
return _new
@property
def _constructor_sliced(self):
def _new(*args, **kwargs):
return Lap(*args, **kwargs).__finalize__(self)
return _new
@property
def base_class_view(self):
"""For a nicer debugging experience; can now view as
dataframe in various IDEs"""
return pd.DataFrame(self)
@cached_property
def telemetry(self):
"""Telemetry data for all laps in `self`
This is a cached (!) property for :meth:`get_telemetry`. It will return the same value as `get_telemetry`
but cache the result so that the involved processing is only done once.
This is mainly provided for convenience and backwards compatibility.
See :meth:`get_telemetry` for more information.
.. note:: Telemetry can only be returned if `self` contains laps of one driver only.
Returns:
instance of :class:`Telemetry`"""
return self.get_telemetry()
[docs] def get_telemetry(self):
"""Telemetry data for all laps in `self`
Telemetry data is the result of merging the returned data from :meth:`get_car_data` and :meth:`get_pos_data`.
This means that telemetry data at least partially contains interpolated values! Telemetry data additionally
already has computed channels added (e.g. Distance).
This method is provided for convenience and compatibility reasons. But using it does usually not produce
the most accurate possible result.
It is recommended to use :meth:`get_car_data` or :meth:`get_pos_data` when possible. This is also faster if
merging of car and position data is not necessary and if not all computed channels are needed.
Resampling during merging is done according to the frequency set by :attr:`TELEMETRY_FREQUENCY`.
.. note:: Telemetry can only be returned if `self` contains laps of one driver only.
Returns:
instance of :class:`Telemetry`
"""
pos_data = self.get_pos_data(pad=1, pad_side='both')
car_data = self.get_car_data(pad=1, pad_side='both')
# calculate driver ahead from from data without padding to
# prevent out of bounds errors
drv_ahead = car_data.iloc[1:-1].add_driver_ahead() \
.loc[:, ('DriverAhead', 'DistanceToDriverAhead',
'Date', 'Time', 'SessionTime')]
car_data = car_data.add_distance().add_relative_distance()
car_data = car_data.merge_channels(drv_ahead)
merged = pos_data.merge_channels(car_data)
return merged.slice_by_lap(self, interpolate_edges=True)
[docs] def get_car_data(self, **kwargs):
"""Car data for all laps in `self`
Slices the car data in :attr:`Session.car_data` using this set of laps and returns the result.
The data returned by this method does not contain computed telemetry channels. The can be added by calling the
appropriate `add_*()` method on the returned telemetry object..
.. note:: Car data can only be returned if `self` contains laps of one driver only.
Args:
**kwargs: Keyword arguments are passed to :meth:`Telemetry.slice_by_lap`
Returns:
instance of :class:`Telemetry`
"""
drv_num = self['DriverNumber'].unique()
if len(drv_num) == 0:
raise ValueError("Cannot slice telemetry because self contains no driver number!")
if len(drv_num) > 1:
raise ValueError("Cannot slice telemetry because self contains Laps of multiple drivers!")
drv_num = drv_num[0]
car_data = self.session.car_data[drv_num].slice_by_lap(self, **kwargs).reset_index(drop=True)
return car_data
[docs] def get_pos_data(self, **kwargs):
"""Pos data for all laps in `self`
Slices the position data in :attr:`Session.pos_data` using this set of laps and returns the result.
.. note:: Position data can only be returned if `self` contains laps of one driver only.
Args:
**kwargs: Keyword arguments are passed to :meth:`Telemetry.slice_by_lap`
Returns:
instance of :class:`Telemetry`
"""
drv_num = self['DriverNumber'].unique()
if len(drv_num) == 0:
raise ValueError("Cannot slice telemetry because self contains no driver number!")
if len(drv_num) > 1:
raise ValueError("Cannot slice telemetry because self contains Laps of multiple drivers!")
drv_num = drv_num[0]
pos_data = self.session.pos_data[drv_num].slice_by_lap(self, **kwargs).reset_index(drop=True)
return pos_data
[docs] def get_weather_data(self):
"""Return weather data for each lap in self.
Weather data is updated once per minute. This means that there are
usually one or two data points per lap. This function will always
return only one data point per lap:
- The first value within the duration of a lap
or
- the last known value before the end of the lap if there are
no values within the duration of a lap
See :func:`fastf1.api.weather_data` for available data
channels.
If you wish to have more control over the data, you can access the
weather data directly in :attr:`Session.weather_data`.
Returns:
pandas.DataFrame
.. doctest::
>>> session = fastf1.get_session(2019, 'Monza', 'Q')
>>> session.load(telemetry=False)
>>> weather_data = session.laps.get_weather_data()
>>> print(weather_data)
Time AirTemp Humidity ... TrackTemp WindDirection WindSpeed
20 0 days 00:20:14.613000 22.5 52.0 ... 35.8 212 2.0
21 0 days 00:21:15.001000 22.5 52.2 ... 36.1 207 2.7
23 0 days 00:23:14.854000 22.7 52.5 ... 37.4 210 2.3
24 0 days 00:24:14.430000 23.2 51.5 ... 37.4 207 3.2
26 0 days 00:26:14.315000 23.6 50.2 ... 37.2 238 1.8
.. ... ... ... ... ... ... ...
36 0 days 00:36:14.426000 23.0 51.1 ... 38.3 192 0.9
37 0 days 00:37:14.391000 23.3 50.0 ... 38.7 213 0.9
28 0 days 00:28:14.324000 23.5 49.9 ... 37.5 183 1.3
34 0 days 00:34:14.385000 23.0 51.7 ... 37.7 272 0.8
35 0 days 00:35:14.460000 23.2 50.3 ... 38.0 339 1.1
<BLANKLINE>
[275 rows x 8 columns]
Joining weather data with lap timing data:
.. doctest::
>>> import pandas as pd # needed additionally to fastf1
# prepare the data for joining
>>> laps = session.laps
>>> laps = laps.reset_index(drop=True)
>>> weather_data = weather_data.reset_index(drop=True)
# exclude the 'Time' column from weather data when joining
>>> joined = pd.concat([laps, weather_data.loc[:, ~(weather_data.columns == 'Time')]], axis=1)
>>> print(joined)
Time DriverNumber ... WindDirection WindSpeed
0 0 days 00:21:01.358000 16 ... 212 2.0
1 0 days 00:22:21.775000 16 ... 207 2.7
2 0 days 00:24:03.991000 16 ... 210 2.3
3 0 days 00:25:24.117000 16 ... 207 3.2
4 0 days 00:27:09.461000 16 ... 238 1.8
.. ... ... ... ... ...
270 0 days 00:36:38.150000 88 ... 192 0.9
271 0 days 00:38:37.508000 88 ... 213 0.9
272 0 days 00:33:27.227000 33 ... 183 1.3
273 0 days 00:35:05.865000 33 ... 272 0.8
274 0 days 00:36:47.787000 33 ... 339 1.1
<BLANKLINE>
[275 rows x 33 columns]
"""
wd = [lap.get_weather_data() for _, lap in self.iterrows()]
if wd:
return pd.concat(wd, axis=1).T
else:
return pd.DataFrame(columns=self.session.weather_data.columns)
[docs] def pick_driver(self, identifier):
"""Return all laps of a specific driver in self based on the driver's
three letters identifier or based on the driver number ::
perez_laps = ff1.pick_driver('PER')
bottas_laps = ff1.pick_driver(77)
kimi_laps = ff1.pick_driver('RAI')
Args:
identifier (str or int): Driver abbreviation or number
Returns:
instance of :class:`Laps`
"""
identifier = str(identifier)
if identifier.isdigit():
return self[self['DriverNumber'] == identifier]
else:
return self[self['Driver'] == identifier]
[docs] def pick_drivers(self, identifiers):
"""Return all laps of the specified drivers in self based on the
drivers' three letters identifier or based on the driver number. This
is the same as :meth:`Laps.pick_driver` but for multiple drivers
at once. ::
some_drivers_laps = ff1.pick_drivers([5, 'BOT', 7])
Args:
identifiers (iterable): Multiple driver abbreviations or driver numbers (can be mixed)
Returns:
instance of :class:`Laps`
"""
names = [n for n in identifiers if not str(n).isdigit()]
numbers = [str(n) for n in identifiers if str(n).isdigit()]
drv, num = self['Driver'], self['DriverNumber']
return self[(drv.isin(names) | num.isin(numbers))]
[docs] def pick_team(self, name):
"""Return all laps of a specific team in self based on the
team's name ::
mercedes = ff1.pick_team('Mercedes')
alfa_romeo = ff1.pick_team('Alfa Romeo')
Have a look to :attr:`fastf1.plotting.TEAM_COLORS` for a quick reference on team names.
Args:
name (str): Team name
Returns:
instance of :class:`Laps`
"""
return self[self['Team'] == name]
[docs] def pick_teams(self, names):
"""Return all laps of the specified teams in self based on the teams'
name. This is the same as :meth:`Laps.pick_team` but for multiple
teams at once. ::
some_drivers_laps = ff1.pick_teams(['Mercedes', 'Williams'])
Args:
names (iterable): Multiple team names
Returns:
instance of :class:`Laps`
"""
return self[self['Team'].isin(names)]
[docs] def pick_fastest(self, only_by_time=False):
"""Return the lap with the fastest lap time.
This method will by default return the quickest lap out of self, that
is also marked as personal best lap of a driver.
If the quickest lap by lap time is not marked as personal best, this
means that it was not counted. This can be the case for example, if
the driver exceeded track limits and the lap time was deleted.
If no lap is marked as personal best lap or self contains no laps,
an empty Lap object will be returned.
The check for personal best lap can be disabled, so that any quickest
lap will be returned.
Args:
only_by_time (bool): Ignore whether any laps are marked as
personal best laps and simply return the lap that has the
lowest lap time.
Returns:
instance of :class:`Lap`
"""
if only_by_time:
laps = self # all laps
else:
# select only laps marked as personal fastest
laps = self.loc[self['IsPersonalBest'] == True] # noqa: E712 comparison with True
if not laps.size:
return Lap(index=self.columns)
lap = laps.loc[laps['LapTime'].idxmin()]
if isinstance(lap, pd.DataFrame):
# More laps, same time
lap = lap.iloc[0] # take first clocked
return lap
[docs] def pick_quicklaps(self, threshold=None):
"""Return all laps with `LapTime` faster than a certain limit. By
default the threshold is 107% of the best `LapTime` of all laps
in self.
Args:
threshold (optional, float): custom threshold coefficent
(e.g. 1.05 for 105%)
Returns:
instance of :class:`Laps`
"""
if threshold is None:
threshold = Laps.QUICKLAP_THRESHOLD
time_threshold = self['LapTime'].min() * threshold
return self[self['LapTime'] < time_threshold]
[docs] def pick_tyre(self, compound):
"""Return all laps in self which were done on a specific compound.
Args:
compound (string): may be "SOFT", "MEDIUM", "HARD", "INTERMEDIATE" or "WET"
Returns:
instance of :class:`Laps`
"""
return self[self['Compound'] == compound]
[docs] def pick_track_status(self, status, how='equals'):
"""Return all laps set under a specific track status.
Args:
status (str): The track status as a string, e.g. '1'
how (str): one of 'equals'/'contains'
For example, if how='equals', status='2' will only match '2'.
If how='contains', status='2' will also match '267' and similar
Returns:
instance of :class:`Laps`
"""
if how == 'equals':
return self[self['TrackStatus'] == status]
elif how == 'contains':
return self[self['TrackStatus'].str.contains(status, regex=False)]
else:
raise ValueError(f"Invalid value '{how}' for kwarg 'how'")
[docs] def pick_wo_box(self):
"""Return all laps which are NOT in laps or out laps.
Returns:
instance of :class:`Laps`
"""
return self[pd.isnull(self['PitInTime']) & pd.isnull(self['PitOutTime'])]
[docs] def pick_accurate(self):
"""Return all laps which pass the accuracy validation check
(lap['IsAccurate'] is True).
Returns:
instance of :class:`Laps`
"""
return self[self['IsAccurate']]
[docs] def iterlaps(self, require=()):
"""Iterator for iterating over all laps in self.
This method wraps :meth:`pandas.DataFrame.iterrows`.
It additionally provides the `require` keyword argument.
Args:
require (optional, iterable): Require is a list of column/telemetry channel names. All names listed in
`require` must exist in the data and have a non-null value (tested with :func:`pandas.is_null`). The
iterator only yields laps for which this is true. If require is left empty, the iterator will yield
all laps.
Yields:
(index, instance of :class:`Lap`)
"""
for index, lap in self.iterrows():
if require:
# make sure that all required values even exist in the index
if any(val not in lap.index.values for val in require):
continue
require = set(require).intersection(set(lap.index.values))
if any(pd.isnull(val) for val in lap.loc[require]):
continue
yield index, lap
[docs]class Lap(pd.Series):
"""Object for accessing lap (timing) data of a single lap.
This class wraps :class:`pandas.Series`. It provides extra functionality for accessing a lap's associated
telemetry data.
"""
_metadata = ['session']
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@property
def _constructor(self):
def _new(*args, **kwargs):
return Lap(*args, **kwargs).__finalize__(self)
return _new
@cached_property
def telemetry(self):
"""Telemetry data for this lap
This is a cached (!) property for :meth:`get_telemetry`. It will return the same value as `get_telemetry`
but cache the result so that the involved processing is only done once.
This is mainly provided for convenience and backwards compatibility.
See :meth:`get_telemetry` for more information.
Returns:
instance of :class:`Telemetry`"""
return self.get_telemetry()
[docs] def get_telemetry(self):
"""Telemetry data for this lap
Telemetry data is the result of merging the returned data from :meth:`get_car_data` and :meth:`get_pos_data`.
This means that telemetry data at least partially contains interpolated values! Telemetry data additionally
already has computed channels added (e.g. Distance).
This method is provided for convenience and compatibility reasons. But using it does usually not produce
the most accurate possible result.
It is recommended to use :meth:`get_car_data` or :meth:`get_pos_data` when possible. This is also faster if
merging of car and position data is not necessary and if not all computed channels are needed.
Resampling during merging is done according to the frequency set by :attr:`TELEMETRY_FREQUENCY`.
Returns:
instance of :class:`Telemetry`
"""
pos_data = self.get_pos_data(pad=1, pad_side='both')
car_data = self.get_car_data(pad=1, pad_side='both')
# calculate driver ahead from from data without padding to
# prevent out of bounds errors
drv_ahead = car_data.iloc[1:-1].add_driver_ahead() \
.loc[:, ('DriverAhead', 'DistanceToDriverAhead',
'Date', 'Time', 'SessionTime')]
car_data = car_data.add_distance().add_relative_distance()
car_data = car_data.merge_channels(drv_ahead)
merged = pos_data.merge_channels(car_data)
return merged.slice_by_lap(self, interpolate_edges=True)
[docs] def get_car_data(self, **kwargs):
"""Car data for this lap
Slices the car data in :attr:`Session.car_data` using this lap and returns the result.
The data returned by this method does not contain computed telemetry channels. The can be added by calling the
appropriate `add_*()` method on the returned telemetry object.
Args:
**kwargs: Keyword arguments are passed to :meth:`Telemetry.slice_by_lap`
Returns:
instance of :class:`Telemetry`
"""
car_data = self.session.car_data[self['DriverNumber']].slice_by_lap(self, **kwargs).reset_index(drop=True)
return car_data
[docs] def get_pos_data(self, **kwargs):
"""Pos data for all laps in `self`
Slices the position data in :attr:`Session.pos_data` using this lap and returns the result.
Args:
**kwargs: Keyword arguments are passed to :meth:`Telemetry.slice_by_lap`
Returns:
instance of :class:`Telemetry`
"""
pos_data = self.session.pos_data[self['DriverNumber']].slice_by_lap(self, **kwargs).reset_index(drop=True)
return pos_data
[docs] def get_weather_data(self):
"""Return weather data for this lap.
Weather data is updated once per minute. This means that there are
usually one or two data points per lap. This function will always
return only one data point:
- The first value within the duration of a lap
or
- the last known value before the end of the lap if there are
no values within the duration of a lap
See :func:`fastf1.api.weather_data` for available data
channels.
If you wish to have more control over the data, you can access the
weather data directly in :attr:`Session.weather_data`.
Returns:
pandas.Series
.. doctest::
>>> session = fastf1.get_session(2019, 'Monza', 'Q')
>>> session.load(telemetry=False)
>>> lap = session.laps.pick_fastest()
>>> lap['LapStartTime']
Timedelta('0 days 01:09:55.561000')
>>> lap.get_weather_data()
Time 0 days 01:10:15.292000
AirTemp 23.0
Humidity 51.9
Pressure 992.4
Rainfall False
TrackTemp 37.8
WindDirection 166
WindSpeed 0.8
Name: 70, dtype: object
"""
# get first value within the duration of the lap
mask = ((self.session.weather_data['Time'] >= self['LapStartTime']) &
(self.session.weather_data['Time'] <= self['Time']))
samples = self.session.weather_data[mask]
if not samples.empty:
return samples.iloc[0]
# fallback: get last value before the lap ended
mask = self.session.weather_data['Time'] <= self['Time']
samples = self.session.weather_data[mask]
if not samples.empty:
return samples.iloc[-1]
# no data: return an empty Series with the correct index names
return pd.Series(index=self.session.weather_data.columns)
[docs]class SessionResults(pd.DataFrame):
"""This class provides driver and result information for all drivers that
participated in a session.
This class subclasses a :class:`pandas.DataFrame` and the usual methods
provided by pandas can be used to work with the data.
**All dataframe columns will always exist even if they are not relevant
for the current session!**
The following information is provided for each driver as a column of the
dataframe:
- ``DriverNumber`` | :class:`str` |
The number associated with this driver in this session (usually the
drivers permanent number)
- ``BroadcastName`` | :class:`str` |
First letter of the drivers first name plus the drivers full last name
in all capital letters. (e.g. 'P GASLY')
- ``FullName`` | :class:`str` |
The drivers full name (e.g. "Pierre Gasly")
- ``Abbreviation`` | :class:`str` |
The drivers three letter abbreviation (e.g. "GAS")
- ``TeamName`` | :class:`str` |
The team name (short version without title sponsors)
- ``TeamColor`` | :class:`str` |
The color commonly associated with this team (hex value)
- ``FirstName`` | :class:`str` |
The drivers first name
- ``LastName`` | :class:`str` |
The drivers last name
- ``Position`` | :class:`float` |
The drivers finishing position (values only given if session is
'Race', 'Qualifying' or 'Sprint Qualifying')
- ``GridPosition`` | :class:`float` |
The drivers starting position (values only given if session is
'Race' or 'Sprint Qualifying')
- ``Q1`` | :class:`pd.Timedelta` |
The drivers best Q1 time (values only given if session is
'Qualifying')
- ``Q2`` | :class:`pd.Timedelta` |
The drivers best Q2 time (values only given if session is
'Qualifying')
- ``Q3`` | :class:`pd.Timedelta` |
The drivers best Q3 time (values only given if session is
'Qualifying')
- ``Time`` | :class:`pd.Timedelta` |
The drivers total race time (values only given if session is
'Race' or 'Sprint Qualifying' and the driver was not more than one
lap behind the leader)
- ``Status`` | :class:`str` |
A status message to indicate if and how the driver finished the race
or to indicate the cause of a DNF. Possible values include but are
not limited to 'Finished', '+ 1 Lap', 'Crash', 'Gearbox', ...
(values only given if session is 'Race' or 'Sprint Qualifying')
- ``Status`` | :class:`float` |
The number of points received by each driver for their finishing
result.
By default, the session results are indexed by driver number and sorted by
finishing position.
.. note:: This class is usually not instantiated directly. You should
create a session and access the session result through the
:attr:`Session.results` property.
Args:
*args: passed on to :class:`pandas.DataFrame` superclass
force_default_cols (bool): Enforce that all default columns and only
the default columns exist
**kwargs: passed on to :class:`pandas.DataFrame` superclass
(except 'columns' which is unsupported for this object)
.. versionadded:: 2.2
"""
_COL_TYPES = {
'DriverNumber': str,
'BroadcastName': str,
'Abbreviation': str,
'TeamName': str,
'TeamColor': str,
'FirstName': str,
'LastName': str,
'FullName': str,
'Position': float,
'GridPosition': float,
'Q1': 'timedelta64[ns]',
'Q2': 'timedelta64[ns]',
'Q3': 'timedelta64[ns]',
'Time': 'timedelta64[ns]',
'Status': str,
'Points': float
}
_internal_names = ['base_class_view']
def __init__(self, *args, force_default_cols=False, **kwargs):
if force_default_cols:
kwargs['columns'] = list(self._COL_TYPES.keys())
super().__init__(*args, **kwargs)
# apply column specific dtypes
for col, _type in self._COL_TYPES.items():
if col not in self.columns:
continue
if self[col].isna().all():
if _type == 'timedelta64[ns]':
self[col] = pd.Series(dtype='timedelta64[ns]')
else:
self[col] = _type()
self[col] = self[col].astype(_type)
def __repr__(self):
return self.base_class_view.__repr__()
@property
def _constructor(self):
def _new(*args, **kwargs):
return SessionResults(*args, **kwargs).__finalize__(self)
return _new
@property
def _constructor_sliced(self):
def _new(*args, **kwargs):
return DriverResult(*args, **kwargs).__finalize__(self)
return _new
@property
def base_class_view(self):
"""For a nicer debugging experience; can view DataFrame through
this property in various IDEs"""
return pd.DataFrame(self)
[docs]class DriverResult(pd.Series):
"""This class provides driver and result information for a single driver.
This class subclasses a :class:`pandas.Series` and the usual methods
provided by pandas can be used to work with the data.
For information on which data is available, see :class:`SessionResult`.
.. note:: This class is usually not instantiated directly. You should
create a session and access the driver result through
:func:`Session.get_driver` or by slicing the session result.
Args:
*args: passed on to :class:`pandas.Series` superclass
**kwargs: passed on to :class:`pandas.Series` superclass
.. versionadded:: 2.2
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._getattr_override = True # TODO: remove in v2.3
@property
def _constructor(self):
def _new(*args, **kwargs):
return DriverResult(*args, **kwargs).__finalize__(self)
return _new
def __getattribute__(self, name):
# TODO: remove in v2.3
if name == 'name' and getattr(self, '_getattr_override', False):
if 'FirstName' in self:
warnings.warn(
"The `Driver.name` property is deprecated and will be"
"removed in a future version.\n"
"Use `Driver['FirstName']` or `Driver.FirstName` instead.",
FutureWarning
)
# name may be accessed by pandas internals to, when data
# does not exist yet
return self['FirstName']
return super().__getattribute__(name)
def __repr__(self):
# don't show .name deprecation message when .name is accessed internally
with warnings.catch_warnings():
warnings.filterwarnings('ignore',
message=r".*property is deprecated.*")
return super().__repr__()
@property
def dnf(self):
"""True if driver did not finish"""
return not (self.Status[3:6] == 'Lap' or self.Status == 'Finished')
@property
def grid(self):
"""Grid position
.. deprecated:: 2.2
Use ``Driver['GridPosition']`` instead
"""
# TODO: remove in v2.3
warnings.warn("The `Driver.grid` property is deprecated and will be"
"removed in a future version.\n"
"Use `Driver['GridPosition']` or `Driver.GridPosition` "
"instead.", FutureWarning)
return self['GridPosition']
@property
def position(self):
"""Finishing position
.. deprecated:: 2.2
Use ``Driver['Position']`` instead
"""
# TODO: remove in v2.3
warnings.warn("The `Driver.position` property is deprecated and will be"
"removed in a future version.\n"
"Use `Driver['Position']` or `Driver.Position` "
"instead.", FutureWarning)
return self['Position']
@property
def familyname(self):
"""Driver family name
.. deprecated:: 2.2
Use ``Driver['LastName']`` instead
"""
# TODO: remove in v2.3
warnings.warn("The `Driver.position` property is deprecated and will be"
"removed in a future version.\n"
"Use `Driver['LastName']` or `Driver.LastName` "
"instead.", FutureWarning)
return self['LastName']
@property
def team(self):
"""Team name
.. deprecated:: 2.2
Use ``Driver['TeamName']`` instead
"""
# TODO: remove in v2.3
warnings.warn("The `Driver.team` property is deprecated and will be"
"removed in a future version.\n"
"Use `Driver['TeamName']` or `Driver.TeamName` "
"instead.", FutureWarning)
return self['TeamName']
class Driver:
"""
.. deprecated:: 2.2
Use :class:`fastf1.core.DriverResult` instead
"""
def __new__(cls, *args, **kwargs):
warnings.warn("`fastf1.core.Driver` has been deprecated and will be"
"removed in a future version.\n"
"Use `fastf1.core.DriverResult` instead.",
FutureWarning)
return DriverResult()
class DataNotLoadedError(Exception):
"""Raised if an attempt is made to access data that has not been loaded
yet."""
pass
[docs]class NoLapDataError(Exception):
"""Raised if the API request does not fail but there is no usable data after processing the result."""
def __init__(self, *args):
super(NoLapDataError, self).__init__("Failed to load session because the API did not provide any usable data.")
[docs]class InvalidSessionError(Exception):
"""Raised if no session for the specified event name, type and year
can be found."""
def __init__(self, *args):
super(InvalidSessionError, self).__init__(
"No matching session can be found."
)