Aswinesag/f1-telemetry-anomaly-engine

# F1 遥测异常引擎 面向生产环境的遥测流水线，用于模拟高频一级方程式（F1）传感器数据流，推导基于物理的热力学特征，执行异步 ML 推理，并在类似维修区墙（pitwall）的仪表盘上可视化异常特征。 当前的实现基于解耦的流式架构： - **Kafka** 负责传输回放的遥测数据帧。 - **FastAPI** 托管健康/就绪状态端点、数据摄取以及 WebSocket 流。 - **InferenceWorker** 在后台执行非阻塞的模型推理。 - **Redis** 存储最新处理的推理结果，以实现低延迟的仪表盘读取。 - **PostgreSQL** 通过 SQLAlchemy 2.0 异步会话持久化遥测快照和异常分数。 - **Next.js** 在端口 `8502` 上渲染维修区墙（pitwall）仪表盘。 # **演示** https://github.com/user-attachments/assets/ed0b3696-9afb-4ec1-83f8-ffb566635091 ## 系统概述 ``` FastF1 / CSV replay | v Telemetry replay producer | v Kafka topic: f1-telemetry-bus | v FastAPI inference-api | +--> async Kafka consumer +--> bounded InferenceWorker queue +--> vectorized PhysicsEngine feature generation +--> PyTorch virtual thermal sensor +--> PyTorch anomaly autoencoder +--> Redis latest-result cache +--> PostgreSQL telemetry_snapshots table | v WebSocket / HTTP latest telemetry | v Next.js pitwall dashboard ``` ## 实现状态 ### 已完成 - 用于 Kafka、ZooKeeper、Redis、PostgreSQL、FastAPI、仪表盘及可选回放的 Docker Compose 编排。 - 具备生命周期管理的 Kafka 消费者、Redis 客户端、DB 引擎和推理 worker 的异步 FastAPI 服务。 - `/health`、`/ready`、`/telemetry`、`/telemetry/latest` 和 `/ws/telemetry` 端点。 - `src/engine/physics.py` 中无状态、可复用的向量化物理引擎。 - 使用 1D CNN + 双向 LSTM 的 PyTorch 虚拟传感器模型。 - 用于异常隔离的 PyTorch 残差自编码器。 - 用于遥测快照的 SQLAlchemy 2.0 异步持久化模型。 - 由 Redis 支持的最新推理结果，用于仪表盘数据填充和 WebSocket 更新。 - Next.js + Tailwind + Recharts 维修区墙（pitwall）仪表盘。 - Docker 运行时依赖拆分： - 用于 API 推理的 `requirements-api.txt`。 - 用于 Kafka 回放的 `requirements-replay.txt`。 - 用于完整本地训练/原型工具的 `requirements.txt`。 - 主机 API 端口移至 `18080`，因为 Windows 通常会保留端口 `8000`。 ### 当前运行端口 | 组件 | 主机 URL / 端口 | 容器端口 | |---|---:|---:| | 仪表盘 | `http://localhost:8502` | `8502` | | 推理 API | `http://localhost:18080` | `8000` | | Kafka 外部监听器 | `localhost:9092` | `9092` | | PostgreSQL | `localhost:5432` | `5432` | | Redis | `localhost:6379` | `6379` | | ZooKeeper | `localhost:2181` | `2181` | ## 仓库结构 ``` . ├── api/ │ ├── main.py # FastAPI app, lifespan, routes, Kafka consumer task │ ├── schemas.py # Pydantic request/response schemas │ ├── worker.py # Async inference worker, Redis latest cache, DB persistence │ └── database/ │ └── models.py # SQLAlchemy 2.0 async engine/session and TelemetrySnapshot model ├── config/ │ └── config.yaml # Sampling rate, feature columns, model hyperparameters, threshold config ├── data/ │ ├── virtual_sensor.pt # Trained virtual thermal sensor checkpoint │ ├── isolation_engine.pt # Trained anomaly autoencoder checkpoint │ └── raw_samples/ # Replay CSV generated by training ├── f1-dashboard/ │ └── app/ │ ├── Dashboard.tsx # Pitwall dashboard UI │ └── page.tsx # Next.js route entry ├── src/ │ ├── engine/ │ │ └── physics.py # Vectorized thermodynamic feature engine │ ├── models/ │ │ ├── virtual_sensor.py # 1D CNN + BiLSTM virtual sensor │ │ └── autoencoder.py # Residual reconstruction anomaly detector │ └── pipeline/ │ ├── etl_processor.py # FastF1 extraction, alignment, scaling │ ├── ingestion.py # Kafka producer/consumer helpers │ └── replay_simulation.py # CSV-to-Kafka replay simulator ├── tests/ │ └── test_physics.py # Physics engine tests ├── docker-compose.yml ├── Dockerfile # Next.js dashboard image ├── Dockerfile.api # FastAPI inference image ├── Dockerfile.replay # Telemetry replay image ├── requirements.txt # Full local training/prototype stack ├── requirements-api.txt # Runtime API dependencies with CPU Torch ├── requirements-replay.txt # Lightweight replay dependencies └── train.py # End-to-end training pipeline ``` ## 技术栈 ### 后端 / 流处理 - **FastAPI** 用于异步 HTTP 和 WebSocket API。 - **Uvicorn** ASGI 运行时。 - **aiokafka** 用于 API 内部的异步 Kafka 消费。 - **kafka-python** 用于回放生产者兼容性。 - **Redis asyncio 客户端** 用于低延迟的最新结果存储。 - **SQLAlchemy 2.0 异步** 搭配 **asyncpg** 用于 PostgreSQL 持久化。 - **Pydantic** 用于类型化的遥测请求契约。 ### 机器学习 / 数据 - **PyTorch** 用于虚拟传感器和异常自编码器。 - **NumPy / Pandas** 用于向量化特征生成。 - **scikit-learn MinMaxScaler** 与虚拟传感器 checkpoint 存储在一起。 - **FastF1** 用于训练流水线中的历史 F1 遥测数据提取。 - **SciPy CubicSpline** 用于训练期间的遥测重采样/对齐。 ### 前端 - **Next.js 16** 搭配 React。 - **Tailwind CSS 4** 用于样式设计。 - **Recharts** 用于响应式遥测图表。 - **lucide-react** 用于仪表盘图标。 ### 基础设施 - **Docker Compose** 用于本地多服务编排。 - **Confluent Kafka + ZooKeeper** 用于流传输。 - **PostgreSQL 16 Alpine** 用于遥测快照存储。 - **Redis 7 Alpine** 用于最新推理缓存。 - API 镜像中使用 **仅支持 CPU 的 PyTorch wheel**，以避免庞大的 CUDA 下载。 ## 核心架构决策 ### 1. 解耦的异步推理 API 不会直接在路由处理器中运行繁重的推理。传入的 Kafka 消息和可选的 HTTP 遥测 POST 请求会被推送到有界的 `asyncio.Queue` 中。`InferenceWorker` 消费队列，保持滚动序列窗口，在后台任务中运行模型推理，并将结果写入 Redis/PostgreSQL。 这可以保持： - HTTP 端点的响应能力。 - WebSocket 流与模型执行相互独立。 - Kafka 摄取与仪表盘读取相互隔离。 - 持久化失败被限制在 worker 日志记录中，而不会导致 WebSocket 崩溃。 ### 2. Redis 作为最新状态缓存 仪表盘通常需要最新的遥测状态，而不是完整的数据库查询。worker 将最新的推理 payload 写入： ``` telemetry:inference:latest ``` 仪表盘从以下位置填充数据： ``` GET /telemetry/latest ``` 然后它从以下位置接收实时更新： ``` ws://localhost:18080/ws/telemetry ``` ### 3. PostgreSQL 用于历史快照 每个处理过的推理结果都可以作为 `TelemetrySnapshot` 行进行持久化，其中包含： - 原始遥测快照。 - 构建的物理快照。 - 预测的热力学状态。 - 实际的合成制动温度目标。 - 异常分数。 - 后端模型阈值。 - 布尔值异常标志。 - 模型版本元数据。 ### 4. 共享的物理引擎 `src/engine/physics.py` 被特意设计为无状态的，并且可以被以下两者导入： - `train.py` / ETL 流水线。 - `api/worker.py` / 实时推理运行时。 这避免了由于重复的特征公式导致的训练与服务偏差。 ### 5. 运行时依赖拆分 原始的完整依赖文件对于运行时容器来说过于臃肿。该项目现在将依赖项拆分为： - `requirements.txt`：本地训练和完整实验。 - `requirements-api.txt`：带有仅支持 CPU Torch 的 API 推理运行时。 - `requirements-replay.txt`：仅用于回放生产者。 ## 物理特征工程 物理引擎使用向量化的 Pandas/NumPy 操作推导热力学和车辆动力学特征： - `Speed_ms` - `Delta_KE` - `Acceleration` - `Longitudinal_G` - `Aero_Drag_N` - `Aero_Downforce_N` - `Effective_Weight_N` - `Brake_Work_EMA` - `Convective_Cooling_Factor` - `Brake_Temp_Target` 所需的原始遥测列： ``` Speed, Brake ``` 配置的模型特征列： ``` raw_channels: - Speed - Throttle - Brake - RPM - nGear physics_engineered: - Delta_KE - Brake_Work_EMA - Acceleration - Longitudinal_G - Aero_Drag_N - Aero_Downforce_N - Effective_Weight_N - Convective_Cooling_Factor ``` ## ML 流水线 ### 虚拟热力学传感器 在 `src/models/virtual_sensor.py` 中实现。 模型结构： ``` Input sequence -> Conv1D feature extractor -> BatchNorm + ReLU + Dropout -> 2-layer bidirectional LSTM -> dense regressor head -> predicted brake/thermal temperature ``` 模型预测的合成目标： ``` Brake_Temp_Target ``` ### 异常隔离引擎 在 `src/models/autoencoder.py` 中实现。 自编码器基于残差进行训练： ``` abs(actual_temperature - predicted_temperature) ``` 在推理期间： ``` anomaly_score = reconstruction_loss(residual) is_anomaly = anomaly_score > alert_threshold ``` 阈值在训练期间根据验证残差重建损失计算得出： ``` mean(losses) + z_score_multiplier * std(losses) ``` 乘数配置于： ``` anomaly_detection: initial_threshold: 3.5 ``` ## API 端点 基础 URL： ``` http://localhost:18080 ``` | 方法 | 路径 | 用途 | |---|---|---| | `GET` | `/health` | 存活检查。即使依赖项未就绪，也会返回 API 状态。 | | `GET` | `/ready` | 针对 Redis、DB、Kafka 任务和 worker 的就绪检查。 | | `POST` | `/telemetry` | 可选的直接遥测摄取，进入 worker 队列。 | | `GET` | `/telemetry/latest` | 从 Redis 获取的最新已处理推理结果。 | | `WS` | `/ws/telemetry` | 实时仪表盘遥测流。 | 遥测 payload 示例： ``` { "TimeSec": 85.2, "Speed": 280.0, "Throttle": 40.0, "Brake": 92.0, "RPM": 11200.0, "nGear": 6 } ``` 推理 payload 示例： ``` { "CapturedAt": "2026-07-12T10:20:30.000000+00:00", "TimeSec": 85.2, "Speed": 280.0, "Brake": 92.0, "Predicted_Temp": 203.4, "Actual_Temp": 220.7, "Anomaly_Score": 294.68, "Alert_Threshold": 40783.31, "Is_Anomaly": false } ``` ## Docker Compose 服务 | 服务 | 用途 | |---|---| | `zookeeper` | Kafka 协调。 | | `kafka` | 遥测事件总线。 | | `postgres` | 异步持久化目标。 | | `redis` | 最新推理缓存。 | | `inference-api` | FastAPI 运行时和模型推理 worker。 | | `pitwall-dashboard` | Next.js 仪表盘。 | | `telemetry-replay` | 用于将 CSV 回放到 Kafka 的可选配置文件服务。 | ## 快速开始 ### 1. 构建并启动核心服务 ``` docker compose up --build ``` 打开： ``` http://localhost:8502 ``` 检查 API： ``` curl http://localhost:18080/health curl http://localhost:18080/ready ``` ### 2. 将遥测数据回放到 Kafka 在核心栈健康运行后： ``` docker compose --profile replay run --rm telemetry-replay ``` 仪表盘只有在遥测数据处理完成后才会显示实时指标。它可以通过 `/telemetry/latest` 从 Redis 填充最近处理的点数据。 ### 3. 停止技术栈 ``` docker compose down ``` 删除 PostgreSQL 卷数据： ``` docker compose down -v ``` ## 本地开发 ### 后端 安装完整的本地依赖： ``` pip install -r requirements.txt ``` 本地运行 API： ``` uvicorn api.main:app --host 0.0.0.0 --port 8000 ``` 如果 Windows 阻止了主机端口 `8000`，请使用： ``` uvicorn api.main:app --host 0.0.0.0 --port 18080 ``` ### 仪表盘 ``` cd f1-dashboard npm install npm run dev -- --hostname 127.0.0.1 -p 8502 ``` 仪表盘期望： ``` NEXT_PUBLIC_WS_URL=ws://localhost:18080/ws/telemetry ``` ### 训练 运行完整的训练流水线： ``` python train.py ``` 这将： 1. 通过 FastF1 获取历史遥测数据。 2. 重采样并对齐遥测数据。 3. 生成物理特征。 4. 训练虚拟热力学传感器。 5. 训练残差自编码器。 6. 保存： - `data/virtual_sensor.pt` - `data/isolation_engine.pt` - `data/raw_samples/monza_ver_cleaned.csv` ### 不使用 Docker 进行回放 首先启动 Kafka，然后运行： ``` python src/pipeline/replay_simulation.py ``` 对于本地主机 Kafka： ``` $env:KAFKA_BOOTSTRAP_SERVERS="localhost:9092" python src/pipeline/replay_simulation.py ``` ## 配置 主要配置文件： ``` config/config.yaml ``` 重要字段： | 字段 | 用途 | |---|---| | `system.target_frequency_hz` | 回放和模型采样频率。 | | `features.raw_channels` | 模型使用的原始遥测列。 | | `features.physics_engineered` | 模型使用的物理特征列。 | | `features.target_channel` | 合成训练目标。 | | `model_hyperparameters.sequence_length` | 滚动推理窗口大小。 | | `model_hyperparameters.hidden_dimension` | LSTM 隐藏层大小。 | | `anomaly_detection.initial_threshold` | 用于阈值生成的 Z-score 乘数。 | ## 环境变量 | 变量 | 默认值 | 使用者 | |---|---|---| | `KAFKA_BOOTSTRAP_SERVERS` | `localhost:9092` | API, 回放 | | `KAFKA_TELEMETRY_TOPIC` | `f1-telemetry-bus` | API | | `KAFKA_CONSUMER_GROUP` | `inference-api` | API | | `DATABASE_URL` | `postgresql+asyncpg://f1_telemetry:f1_telemetry@localhost:5432/f1_telemetry` | API | | `REDIS_URL` | `redis://localhost:6379/0` | API | | `WEBSOCKET_POLL_SECONDS` | `0.05` | API | | `INFERENCE_QUEUE_MAX_SIZE` | `2000` | API | | `MODEL_VERSION` | `unknown` | API 持久化 | | `NEXT_PUBLIC_WS_URL` | `ws://localhost:18080/ws/telemetry` | 仪表盘 | ## 验证 后端语法： ``` python -m py_compile api/main.py api/worker.py api/database/models.py api/schemas.py ``` 物理测试： ``` python -m pytest tests/test_physics.py -q ``` Docker Compose 验证： ``` docker compose config --quiet ``` 仪表盘： ``` cd f1-dashboard npm run lint npm run build ``` ## 运维说明 ### Windows 端口 `8000` 某些 Windows 设置会在排除的 TCP 范围内预留端口 `8000`。该项目在主机端口 `18080` 上公开 API，同时保持容器端口为 `8000`。 检查排除的范围： ``` netsh interface ipv4 show excludedportrange protocol=tcp ``` ### Docker 镜像下载 API 镜像通过 `requirements-api.txt` 使用仅支持 CPU 的 PyTorch，以避免庞大的 CUDA 下载。如果在 Torch 下载时停滞，请检查到以下地址的网络/DNS 访问： ``` download.pytorch.org ``` ### 仪表盘数据行为 仪表盘仅推理数据后才会进行渲染。要生成数据： 1. 启动 Compose。 2. 等待 Kafka/API 就绪。 3. 运行回放配置文件。 4. 打开 `http://localhost:8502`。 ### Git 卫生 生成的文件夹和本地缓存会被忽略： - `__pycache__/` - `.pytest_cache/` - `node_modules/` - `.next/` - `data/f1_cache/` - SQLite 缓存文件 模型 checkpoint 和回放 CSV 默认不会被忽略，因为它们是当前可运行原型状态的一部分。 ## 许可证 本项目基于 MIT 许可证授权。详见 `LICENSE.md`。

标签：Apex, AV绕过, F1赛车遥测, FastAPI, PyTorch, 凭据扫描, 实时数据流, 异常检测, 搜索引擎查询, 机器学习, 测试用例, 请求拦截, 逆向工具