FastF1 v2.1.1 实时时序数据记录与回放:Live Timing Data 完整实战指南
【免费下载链接】Fast-F1FastF1 is a python package for accessing and analyzing Formula 1 results, schedules, timing data and telemetry项目地址: https://gitcode.com/GitHub_Trending/fa/Fast-F1
FastF1 v2.1.1 引入了两项影响深远的能力:一是可以在比赛进行时通过 SignalR 协议记录 F1 官方实时时序数据流,二是在赛后把这份本地记录作为数据源回放,让那些赛后无法通过 API 获取的数据得以分析与使用。与此同时,本版本还包含一个需要注意的破坏性变更:Session.load_laps默认不再加载遥测数据。本文以 v2.1.1 变更记录(docs/changelog/v2.1.1.rst)为骨架,结合 docs/api_reference/livetiming.rst 的用法文档与fastf1/livetiming/模块源码,完整讲解记录、回放、认证、缓存与排障的每一个环节。
一、v2.1.1 核心变更总览
v2.1.1 版本记录的核心内容可归纳为三部分:
| 变更类型 | 内容 | 影响 |
|---|---|---|
| 新功能 | 支持记录(record)实时时序数据 | 比赛中通过 SignalR 客户端把官方流保存为本地文本文件 |
| 新功能 | 支持使用(use)已记录的实时时序数据作为数据源 | 赛后通过LiveTimingData对象把本地记录喂给Session.load |
| 潜在破坏性变更 | fastf1.Session.load_laps默认不再加载遥测数据,仅加载时序数据 | 需显式要求才会读取遥测,调用方式需相应调整 |
官方给出的变更说明(docs/changelog/v2.1.1.rst)非常简练,但背后对应着一整套fastf1/livetiming/模块(客户端、数据解析、命令行入口)以及Session.load的livedata参数链路(fastf1/core.py)。下面逐一展开。
二、为什么要自己记录实时时序数据
F1 官方在比赛期间通过 SignalR 协议(wss://livetiming.formula1.com/signalrcore)实时推送时序数据。这些数据的特点是:
- 只在直播窗口内可得:部分时序数据(如逐圈
TimingData、车队/车手信息、实时位置流)在赛后可能无法从普通 API 完整获取,或获取范围受限; - 内容远超赛后静态结果:实时流中包含了
TimingAppData、TrackStatus、SessionStatus、WeatherData、RaceControlMessages、Position.z、CarData.z、LapCount、TopThree等二十余类主题,覆盖比赛进程的几乎全部动态细节。
因此,在比赛进行时把它记录下来,相当于拥有了与比赛同步的"原始数据底稿",赛后即可离线回放。FastF1 v2.1.1 提供的正是这条"记录 → 回放"的完整链路:
F1 官方实时流(SignalR) │ SignalRClient 订阅并落盘 ▼ 本地文本文件(原始消息) │ LiveTimingData 解析 ▼ Session.load(livedata=...) → 时序/遥测/圈速等数据模块的类注释也明确说明:记录的数据不能用于实时处理,只能在赛后通过Session.load配合LiveTimingData对象回放(见 fastf1/livetiming/client.py)。
前置条件:实时时序数据属于 F1TV 付费权益。使用该功能需要有效的 F1TV Access/Pro/Premium 订阅,FastF1 会在未登录时引导你完成账户认证(认证细节见 fastf1/internals/f1auth.py)。另外,由于登录方式依赖本地回调,实时时序客户端无法在 Google Colab 这类托管环境或 WebAssembly(Jupyter Lite)环境中使用。
三、记录实时时序数据:SignalR 客户端
记录功能由fastf1.livetiming.client.SignalRClient实现(fastf1/livetiming/client.py)。它有两种使用方式:命令行与 Python 脚本。
3.1 方式一:命令行记录(推荐)
模块提供了命令行入口python -m fastf1.livetiming,解析逻辑见 fastf1/livetiming/main.py:
python -m fastf1.livetiming save saved_data.txt执行后客户端会连接官方流、订阅全部主题并把数据持续写入saved_data.txt,直到超时或手动中断(Ctrl+C)为止。
3.2 方式二:Python 脚本记录
在脚本中创建SignalRClient实例并调用start():
from fastf1.livetiming.client import SignalRClient client = SignalRClient(filename="saved_data.txt") client.start()3.3 参数说明
SignalRClient的完整参数及语义如下(与 docs/api_reference/livetiming.rst 及源码 docstring 一致):
| 参数 | 默认值 | 说明 |
|---|---|---|
filename | (必填) | 输出文件路径 |
filemode | "w" | "w"覆盖写入;"a"追加写入。会话中途重启客户端时用追加模式很有用 |
debug | False | 原用于保存完整 SignalR 消息而非仅数据部分。注意:从当前仓库源码看,该模式已被移除,传入debug=True会直接抛出ValueError("Debug mode is no longer supported.")(fastf1/livetiming/client.py) |
timeout | 60 | 连续多少秒收不到数据则自动退出;设为0可禁用超时 |
logger | None | 默认错误输出到控制台;可传入logging.Logger自定义日志 |
no_auth | False | 设为True时尝试免认证连接,但可能只对部分会话有效,或只能拿到空/不完整数据 |
命令行对应关系:
usage: python -m fastf1.livetiming save [-h] [--append] [--debug] [--timeout TIMEOUT] file positional arguments: file 输出文件名 optional arguments: -h, --help 显示帮助并退出 --append 追加到输出文件;默认覆盖已存在文件 --debug 启用调试模式:保存完整 SignalR 消息而非仅数据 --timeout TIMEOUT 收不到数据后自动退出的超时秒数(默认 60)3.4 底层工作原理
从源码(fastf1/livetiming/client.py)可以看清客户端的关键调用链:
- 预协商(Pre-negotiate):
_run()首先向https://livetiming.formula1.com/signalrcore/negotiate发起OPTIONS请求,拿到AWSALBCORSCookie 并写入请求头(fastf1/livetiming/client.py); - 建立连接:通过
HubConnectionBuilder构建 SignalR Core 连接,access_token_factory指向认证函数get_auth_token(no_auth=True时为None),并注册on_open/on_close/on("feed", ...)回调(fastf1/livetiming/client.py); - 订阅主题:连接建立后发送
Subscribe消息,订阅列表包含Heartbeat、DriverList、TimingData、CarData.z、Position.z、TrackStatus、WeatherData、SessionStatus、LapCount、TopThree等 21 个主题(fastf1/livetiming/client.py); - 落盘:每条消息经
_on_message序列化后写入文件并立即flush()(fastf1/livetiming/client.py); - 监督退出:
_supervise()每秒检查一次最近收到消息的时间,超过timeout即告警并自动关闭连接(fastf1/livetiming/client.py)。
需要注意,当前仓库源码中async_start()已不再提供(会抛出NotImplementedError),因为客户端不再基于 asyncio,统一使用同步的.start()(fastf1/livetiming/client.py)。
四、把已记录数据用作数据源:LiveTimingData
记录完成后,数据默认以"原始文本"形式保存(每行一个 JSON 数组元素,形如['TimingAppData', {...}, '2021-03-27T12:00:32.086Z'],分别对应分类名、消息体、UTC 时间戳)。要回放它,需要借助fastf1.livetiming.data.LiveTimingData(fastf1/livetiming/data.py)。
4.1 基本用法
import fastf1 from fastf1.livetiming.data import LiveTimingData livedata = LiveTimingData('saved_data.txt') session = fastf1.get_testing_session(2021, 1, 1) session.load(livedata=livedata)Session.load的livedata关键字参数会贯穿整条加载链路:_load_session_info、_load_drivers_results、_load_session_status_data、_load_total_lap_count、_load_track_status_data、_load_laps_data、_load_telemetry、_load_weather_data、_load_race_control_messages等全部数据加载方法都接受并透传该对象(见 fastf1/core.py)。也就是说,只要传入livedata,时序数据、圈速、遥测(若可用)、天气、赛道状态、比赛控制消息等都可以从本地记录解析,而不是请求赛后 API。
4.2 多文件与重叠去重
如果一次录制被拆成了多个文件(例如因断连而分两次录制),可以直接传入多个文件名,文件需按时间先后顺序排列:
livedata = LiveTimingData('saved_data_1.txt', 'saved_data_2.txt')多个文件之间允许重叠:load()在解析时会加载"当前文件 + 下一个文件首行",遇到与下一文件首行相同的行即停止解析当前文件,从而自动识别并去除重叠区间的重复数据(fastf1/livetiming/data.py)。这个行为有专门测试验证:test_duplicate_removal构造了两个内容完全相同的临时文件,加载后断言数据量只有 1 份(fastf1/tests/test_livetiming.py)。
早期版本的remove_duplicates参数已废弃——重复数据现在总是会被移除,传入该参数只会触发警告(fastf1/livetiming/data.py)。
4.3 底层解析逻辑
LiveTimingData内部做了三件关键的事:
- 时间基准校准:解析首个文件时,先扫描内容寻找
SessionStatus中状态为Started的消息,以其Utc时间作为会话开始时间(_start_date),随后所有消息时间都转换为相对会话开始的timedelta(fastf1/livetiming/data.py)。若找不到Started记录,则退而使用第一条数据的时间戳作为基准; - 非标准 JSON 修复:F1 官方推送的数据不是严格 JSON(使用单引号、
True/False字面量),解析前会统一替换为合法 JSON('→"、True→true、False→false)(fastf1/livetiming/data.py); - 按分类归档:解析后的
[时间增量, 消息体]按分类名存入self.data字典(fastf1/livetiming/data.py),并提供get(name)、has(name)、list_categories()三个查询接口——首次调用时自动触发load(),因此会有一次性的解析耗时(fastf1/livetiming/data.py)。
即使文件中混入大量坏数据也不会崩溃:test_file_loading_w_errors专门用带大量错误行的参考数据验证了解析的健壮性(fastf1/tests/test_livetiming.py,参考数据位于 fastf1/testing/reference_data/livedata/)。
五、破坏性变更:load_laps 默认不再加载遥测
v2.1.1 明确指出一个可能破坏现有代码的变更:
fastf1.Session.load_laps:数据现在默认在不加载遥测的情况下载入,即只加载时序数据。遥测数据通常本来也不可用,因此这可以避免一个令人困惑的错误。
也就是说,在 v2.1.1 及之后,调用load_laps()时不会再顺带拉取每圈的遥测数据。如果你的分析流程依赖Lap.telemetry之类的字段,需要显式开启遥测加载,而不能假设load_laps会自动带出。
顺带一提,Session.load本身仍然支持灵活的按需加载开关laps / telemetry / weather / messages,并可配合livedata同时指定本地数据源(fastf1/core.py)。在回放录制数据时,建议显式指定所需数据种类,避免加载不必要的部分拖慢首次解析。
六、实战注意事项(官方经验总结)
在正式投入录制前,请记住以下来自官方文档(docs/api_reference/livetiming.rst)的硬性建议:
尽量录制完整会话:录制可能需要提前到比赛开始前 1 小时启动,才能覆盖全部数据。如果会话开头缺失,API 解析器可能无法正确处理数据——缺多少、缺什么将直接影响可用性;
不要混用数据源:同一场比赛,不要既用录制数据又用赛后 API 数据,两者时间基准可能无法正确对齐;
一定要启用缓存:录制数据配合缓存可以大幅加快第二次及之后的加载速度:
fastf1.Cache.enable_cache('path/to/cache/directory')同一场比赛的不同数据源必须使用不同缓存目录:缓存无法区分数据来源。如果同一场比赛已有 API 数据缓存,就不会自动用录制数据重新加载。修改输入源后还需强制刷新缓存一次:
fastf1.Cache.enable_cache('path/to/cache/directory', force_renew=True)注意
force_renew=True只需要在修改输入(例如新增了第二个录制文件)后执行一次;连接约 2 小时后会被服务器断开:官方流似乎会在录制约 2 小时后主动终止连接。若不想有录制空洞,需要在断开前手动启动第二次录制,并使用不同的输出文件名。之后把文件按时间顺序传入
LiveTimingData即可,重叠部分会被自动去重。
七、认证流程与限制
实时时序数据需要 F1TV 订阅认证。认证实现在 fastf1/internals/f1auth.py 中,流程大致是:
- 客户端在本地启动一个临时 HTTP 服务,并打印一个登录链接(
f1login.fastf1.dev?port=<port>); - 用户在浏览器中完成 F1TV 账户登录;
- 回调把
subscriptionToken写回本地服务,FastF1 通过jwt库、以 JWKS 公钥(https://api.formula1.com/static/jwks.json)校验令牌签名(RS256); - 校验通过的令牌会被持久化到平台用户数据目录下的
f1auth.json,后续直接复用;令牌失效时会提示重新认证。
因此,该登录方式依赖"本地起服务 + 浏览器跳转",这决定了它无法在 Google Colab、Jupyter Lite 等托管/WebAssembly 环境中运行,只适合在本地开发机上使用。
八、如何验证与继续深入
阅读模块文档:docs/api_reference/livetiming.rst 是这份功能的完整使用手册;
阅读源码:fastf1/livetiming/client.py(客户端)、fastf1/livetiming/data.py(数据对象)、fastf1/livetiming/main.py(CLI 入口);
运行测试:仓库内置了录制数据的解析与回放测试,可直接执行验证:
pytest fastf1/tests/test_livetiming.py测试用参考数据位于 fastf1/testing/reference_data/livedata/,其中
2021_1_FP3.txt是一份真实格式的录制样例,with_errors.txt用于验证容错性;若要了解认证细节,见 fastf1/internals/f1auth.py 及文档 docs/api_reference/accounts_auth.rst。
结语
v2.1.1 是 FastF1 数据能力的一次重要补全:它把"只有直播时才能看到的官方时序流"变成了可保存、可离线回放的本地资产,同时通过load_laps默认行为的调整让圈速数据的加载更符合直觉、避免无意义的遥测请求报错。只要遵循"完整录制、分开缓存、按时间序传文件"这三条原则,你就能稳定地把任何一场比赛的实时时序数据完整归档,并在赛后用 FastF1 的完整分析管线任意挖掘。
【免费下载链接】Fast-F1FastF1 is a python package for accessing and analyzing Formula 1 results, schedules, timing data and telemetry项目地址: https://gitcode.com/GitHub_Trending/fa/Fast-F1
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考