驾校训练车违规提醒，压线，超速，实时提醒，辅助教学，输出纠错。
驾校训练车智能违规提醒与辅助教学系统
一、项目概述
1.1 实际应用场景
在驾校科目二和科目三的训练过程中，学员由于操作不熟练，经常会出现压线、超速、熄火、未打转向灯等违规行为。传统模式下，教练需要时刻观察学员操作，口头提醒，但存在以下问题：
典型场景：某驾校在科目二训练中，学员小王在倒库练习时，右后轮多次压到库位边线，但教练未能及时发现并纠正，导致形成错误肌肉记忆，考试时因此挂科。同时，在坡道定点停车时，学员因紧张导致起步溜车，若非教练及时踩副刹，可能发生事故。
1.2 行业痛点
痛点 具体表现 影响
监管盲区 教练无法同时关注多辆训练车 违规操作未被及时纠正
反应滞后 人工观察-判断-提醒存在延迟 错误动作已固化
标准不一 不同教练评判标准存在差异 教学质量不稳定
数据缺失 缺乏训练过程量化记录 难以针对性改进
安全隐患 紧急情况响应不及时 潜在安全风险
二、核心逻辑讲解
2.1 系统架构
┌─────────────────────────────────────────────────────────────────┐
│ 驾校训练车智能违规提醒与辅助教学系统 │
├─────────────┬─────────────┬─────────────┬─────────────┬───────────┤
│ 数据采集层 │ 感知分析层 │ 决策预警层 │ 反馈执行层 │ 数据管理层 │
├─────────────┼─────────────┼─────────────┼─────────────┼───────────┤
│ GPS定位模块 │ 压线检测算法 │ 规则引擎 │ 语音播报 │ 训练日志 │
│ IMU惯性测量 │ 速度监控 │ 风险评估 │ 灯光提示 │ 违规统计 │
│ 摄像头视觉 │ 姿态估计 │ 个性化建议 │ 震动提醒 │ 进步分析 │
│ CAN总线数据 │ 行为识别 │ 教学策略 │ 远程干预 │ 报告生成 │
└─────────────┴─────────────┴─────────────┴─────────────┴───────────┘
2.2 核心算法原理
2.2.1 压线检测算法
电子围栏法：
- 在训练场地关键区域（如库位、边线、坡道）建立虚拟电子围栏
- 使用GPS+IMU融合定位，精度达到厘米级
- 实时计算车辆与边线的距离，当距离小于阈值时触发压线报警
距离计算公式：
d = \frac{|Ax + By + C|}{\sqrt{A^2 + B^2}}
其中 (A,B,C) 为边线方程参数， (x,y) 为车辆位置。
2.2.2 速度监控与超速检测
多源速度融合：
- GPS速度：宏观速度，适合长距离监控
- 轮速传感器：微观速度，适合精确控制
- IMU积分：补充短时间内的速度变化
速度超限判断：
v_{actual} > v_{limit} \times (1 + \alpha)
其中 \alpha 为允许的速度波动容差（通常5%）。
2.2.3 实时纠错与辅助教学
错误分类与建议生成：
- 立即错误：如压线、超速，需立即纠正
- 潜在风险：如车速过慢、方向不稳，需提前预警
- 操作建议：基于错误类型，提供具体改进方法
个性化学习策略：
- 根据学员历史错误，调整提醒频率和方式
- 针对薄弱环节，增加专项训练建议
- 记录进步曲线，激励学员学习
三、项目结构
driving_school_assistant/
├── README.md # 项目说明文档
├── requirements.txt # 依赖包列表
├── main.py # 主程序入口
├── config/ # 配置文件目录
│ ├── config.yaml # 系统配置参数
│ └── training_areas.yaml # 训练场地电子围栏配置
├── core/ # 核心算法模块
│ ├── __init__.py
│ ├── data_acquisition.py # 数据采集模块
│ ├── line_detection.py # 压线检测算法
│ ├── speed_monitor.py # 速度监控模块
│ ├── violation_detector.py # 违规检测器
│ └── teaching_assistant.py # 辅助教学模块
├── hardware/ # 硬件接口模块
│ ├── __init__.py
│ ├── gps_interface.py # GPS设备接口
│ ├── imu_interface.py # IMU设备接口
│ └── can_interface.py # CAN总线接口
├── feedback/ # 反馈执行模块
│ ├── __init__.py
│ ├── voice_prompt.py # 语音播报
│ ├── light_indicator.py # 灯光指示
│ └── remote_control.py # 远程干预
├── data/ # 数据存储
│ ├── __init__.py
│ ├── database.py # 数据库操作
│ └── logger.py # 日志记录
├── visualization/ # 可视化模块
│ ├── __init__.py
│ └── dashboard.py # 实时监控面板
└── utils/ # 工具函数
├── __init__.py
└── helpers.py # 辅助函数
四、核心代码实现
4.1 数据采集模块 (core/data_acquisition.py)
"""
数据采集模块
负责从各类传感器获取车辆训练数据
包括GPS、IMU、CAN总线等多源数据融合
"""
import time
import threading
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from abc import ABC, abstractmethod
import queue
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class VehicleState:
"""车辆状态数据类"""
timestamp: float = 0.0 # 时间戳
latitude: float = 0.0 # 纬度
longitude: float = 0.0 # 经度
altitude: float = 0.0 # 海拔
speed_gps: float = 0.0 # GPS速度 (m/s)
speed_wheel: float = 0.0 # 轮速 (m/s)
acceleration_x: float = 0.0 # X轴加速度
acceleration_y: float = 0.0 # Y轴加速度
acceleration_z: float = 0.0 # Z轴加速度
angular_velocity_x: float = 0.0 # X轴角速度
angular_velocity_y: float = 0.0 # Y轴角速度
angular_velocity_z: float = 0.0 # Z轴角速度
heading: float = 0.0 # 航向角 (度)
steering_angle: float = 0.0 # 方向盘转角
brake_pedal: float = 0.0 # 刹车踏板位置 (0-100%)
accelerator_pedal: float = 0.0 # 油门踏板位置 (0-100%)
gear_position: int = 0 # 档位
engine_rpm: int = 0 # 发动机转速
turn_signal_left: bool = False # 左转向灯
turn_signal_right: bool = False # 右转向灯
emergency_brake: bool = False # 紧急刹车状态
@dataclass
class TrainingArea:
"""训练区域配置类"""
area_id: str # 区域ID
area_name: str # 区域名称
boundary_points: List[tuple] # 边界点列表 [(lat, lon), ...]
speed_limit: float = 10.0 # 速度限制 (m/s)
line_tolerance: float = 0.3 # 压线容差 (m)
is_active: bool = True # 是否激活
class DataSource(ABC):
"""数据源抽象基类"""
@abstractmethod
def connect(self) -> bool:
"""连接数据源"""
pass
@abstractmethod
def disconnect(self) -> None:
"""断开连接"""
pass
@abstractmethod
def read_data(self) -> Dict:
"""读取数据"""
pass
@abstractmethod
def is_connected(self) -> bool:
"""检查连接状态"""
pass
class GPSDataSource(DataSource):
"""
GPS数据源实现
模拟GPS设备数据读取
实际应用中需替换为真实GPS设备接口
"""
def __init__(self, device_id: str = "GPS001", port: str = "/dev/ttyUSB0"):
"""
初始化GPS数据源
Args:
device_id: 设备标识符
port: 串口端口
"""
self.device_id = device_id
self.port = port
self.connected = False
self.mock_mode = True # 模拟模式开关
self.current_position = (39.9042, 116.4074) # 默认北京天安门
self.current_speed = 0.0
self.current_heading = 0.0
def connect(self) -> bool:
"""连接GPS设备"""
try:
if self.mock_mode:
self.connected = True
logger.info(f"GPS设备 {self.device_id} 已连接（模拟模式）")
else:
# 实际GPS设备连接逻辑
# import serial
# self.serial_port = serial.Serial(self.port, 9600, timeout=1)
self.connected = True
logger.info(f"GPS设备 {self.device_id} 已连接（真实模式）")
return True
except Exception as e:
logger.error(f"GPS设备连接失败: {e}")
return False
def disconnect(self) -> None:
"""断开GPS连接"""
self.connected = False
logger.info(f"GPS设备 {self.device_id} 已断开")
def read_data(self) -> Dict:
"""
读取GPS数据
Returns:
GPS数据字典
"""
if not self.connected:
return {}
if self.mock_mode:
# 模拟GPS数据变化
self._update_mock_position()
return {
'latitude': self.current_position[0],
'longitude': self.current_position[1],
'altitude': 50.0,
'speed': self.current_speed,
'heading': self.current_heading,
'timestamp': time.time()
}
def _update_mock_position(self) -> None:
"""更新模拟位置（用于测试）"""
# 模拟车辆缓慢移动
import random
self.current_position = (
self.current_position[0] + random.uniform(-0.00001, 0.00001),
self.current_position[1] + random.uniform(-0.00001, 0.00001)
)
self.current_speed = max(0, min(15, self.current_speed + random.uniform(-0.5, 0.5)))
self.current_heading = (self.current_heading + random.uniform(-5, 5)) % 360
def is_connected(self) -> bool:
return self.connected
class IMUDataSource(DataSource):
"""
IMU数据源实现
获取车辆加速度和角速度数据
"""
def __init__(self, device_id: str = "IMU001"):
"""
初始化IMU数据源
Args:
device_id: 设备标识符
"""
self.device_id = device_id
self.connected = False
self.mock_mode = True
self.acceleration = [0.0, 0.0, 0.0]
self.gyroscope = [0.0, 0.0, 0.0]
def connect(self) -> bool:
"""连接IMU设备"""
try:
if self.mock_mode:
self.connected = True
logger.info(f"IMU设备 {self.device_id} 已连接（模拟模式）")
else:
# 实际IMU设备连接逻辑
self.connected = True
logger.info(f"IMU设备 {self.device_id} 已连接（真实模式）")
return True
except Exception as e:
logger.error(f"IMU设备连接失败: {e}")
return False
def disconnect(self) -> None:
"""断开IMU连接"""
self.connected = False
logger.info(f"IMU设备 {self.device_id} 已断开")
def read_data(self) -> Dict:
"""
读取IMU数据
Returns:
IMU数据字典
"""
if not self.connected:
return {}
if self.mock_mode:
self._update_mock_imu()
return {
'acceleration_x': self.acceleration[0],
'acceleration_y': self.acceleration[1],
'acceleration_z': self.acceleration[2],
'gyro_x': self.gyroscope[0],
'gyro_y': self.gyroscope[1],
'gyro_z': self.gyroscope[2],
'timestamp': time.time()
}
def _update_mock_imu(self) -> None:
"""更新模拟IMU数据"""
import random
# 模拟车辆静止时的微小振动
self.acceleration = [
random.uniform(-0.1, 0.1),
random.uniform(-0.1, 0.1),
-9.8 + random.uniform(-0.2, 0.2) # 重力
]
self.gyroscope = [
random.uniform(-0.5, 0.5),
random.uniform(-0.5, 0.5),
random.uniform(-0.5, 0.5)
]
def is_connected(self) -> bool:
return self.connected
class CANBusDataSource(DataSource):
"""
CAN总线数据源实现
获取车辆CAN总线数据（车速、转向角、踏板位置等）
"""
def __init__(self, device_id: str = "CAN001", channel: str = "can0"):
"""
初始化CAN数据源
Args:
device_id: 设备标识符
channel: CAN通道
"""
self.device_id = device_id
self.channel = channel
self.connected = False
self.mock_mode = True
self.can_data = {
'speed_wheel': 0.0,
'steering_angle': 0.0,
'brake_pedal': 0.0,
'accelerator_pedal': 0.0,
'gear_position': 0,
'engine_rpm': 0,
'turn_signal_left': False,
'turn_signal_right': False,
'emergency_brake': False
}
def connect(self) -> bool:
"""连接CAN总线"""
try:
if self.mock_mode:
self.connected = True
logger.info(f"CAN设备 {self.device_id} 已连接（模拟模式）")
else:
# 实际CAN设备连接逻辑
# import can
# self.bus = can.interface.Bus(channel=self.channel, bustype='socketcan')
self.connected = True
logger.info(f"CAN设备 {self.device_id} 已连接（真实模式）")
return True
except Exception as e:
logger.error(f"CAN设备连接失败: {e}")
return False
def disconnect(self) -> None:
"""断开CAN连接"""
self.connected = False
logger.info(f"CAN设备 {self.device_id} 已断开")
def read_data(self) -> Dict:
"""
读取CAN数据
Returns:
CAN数据字典
"""
if not self.connected:
return {}
if self.mock_mode:
self._update_mock_can_data()
return self.can_data.copy()
def _update_mock_can_data(self) -> None:
"""更新模拟CAN数据"""
import random
self.can_data.update({
'speed_wheel': max(0, self.can_data.get('speed_wheel', 0) + random.uniform(-0.3, 0.3)),
'steering_angle': max(-540, min(540, self.can_data.get('steering_angle', 0) + random.uniform(-2, 2))),
'brake_pedal': max(0, min(100, self.can_data.get('brake_pedal', 0) + random.uniform(-5, 5))),
'accelerator_pedal': max(0, min(100, self.can_data.get('accelerator_pedal', 0) + random.uniform(-3, 3))),
'engine_rpm': max(0, self.can_data.get('engine_rpm', 800) + random.randint(-50, 50)),
'turn_signal_left': random.random() < 0.02,
'turn_signal_right': random.random() < 0.02,
'emergency_brake': random.random() < 0.005
})
def send_command(self, command: Dict) -> bool:
"""
发送CAN命令（用于远程干预）
Args:
command: 命令字典
Returns:
是否发送成功
"""
if not self.connected:
return False
logger.info(f"发送CAN命令: {command}")
return True
def is_connected(self) -> bool:
return self.connected
class DataAcquisitionManager:
"""
数据采集管理器
统一管理多源数据采集，进行数据融合和时间同步
"""
def __init__(self, config: Dict = None):
"""
初始化数据采集管理器
Args:
config: 配置字典
"""
self.config = config or {}
self.data_sources: Dict[str, DataSource] = {}
self.data_queue = queue.Queue(maxsize=1000)
self.is_running = False
self.acquisition_thread: Optional[threading.Thread] = None
self.fusion_callback: Optional[Callable] = None
# 数据缓存
self.latest_gps_data = {}
self.latest_imu_data = {}
self.latest_can_data = {}
def register_data_source(self, name: str, source: DataSource) -> None:
"""
注册数据源
Args:
name: 数据源名称
source: 数据源对象
"""
self.data_sources[name] = source
logger.info(f"已注册数据源: {name}")
def connect_all(self) -> bool:
"""
连接所有数据源
Returns:
是否全部连接成功
"""
success = True
for name, source in self.data_sources.items():
if not source.connect():
logger.error(f"数据源 {name} 连接失败")
success = False
return success
def disconnect_all(self) -> None:
"""断开所有数据源连接"""
for name, source in self.data_sources.items():
source.disconnect()
logger.info("所有数据源已断开")
def start_acquisition(self, fusion_callback: Callable = None) -> None:
"""
启动数据采集
Args:
fusion_callback: 数据融合回调函数
"""
self.fusion_callback = fusion_callback
self.is_running = True
self.acquisition_thread = threading.Thread(target=self._acquisition_loop)
self.acquisition_thread.daemon = True
self.acquisition_thread.start()
logger.info("数据采集已启动")
def stop_acquisition(self) -> None:
"""停止数据采集"""
self.is_running = False
if self.acquisition_thread:
self.acquisition_thread.join(timeout=2.0)
logger.info("数据采集已停止")
def _acquisition_loop(self) -> None:
"""数据采集主循环"""
while self.is_running:
try:
# 读取各数据源数据
gps_data = self.data_sources.get('gps', GPSDataSource()).read_data()
imu_data = self.data_sources.get('imu', IMUDataSource()).read_data()
can_data = self.data_sources.get('can', CANBusDataSource()).read_data()
# 更新缓存
if gps_data:
self.latest_gps_data = gps_data
if imu_data:
self.latest_imu_data = imu_data
if can_data:
self.latest_can_data = can_data
# 数据融合
fused_data = self._fuse_data(gps_data, imu_data, can_data)
# 放入队列
if not self.data_queue.full():
self.data_queue.put(fused_data)
# 调用回调函数
if self.fusion_callback and fused_data:
self.fusion_callback(fused_data)
# 控制采集频率
time.sleep(0.05) # 20Hz
except Exception as e:
logger.error(f"数据采集异常: {e}")
time.sleep(0.1)
def _fuse_data(self, gps_data: Dict, imu_data: Dict, can_data: Dict) -> Optional[VehicleState]:
"""
数据融合，生成统一的车辆状态
Args:
gps_data: GPS数据
imu_data: IMU数据
can_data: CAN数据
Returns:
融合后的车辆状态
"""
if not any([gps_data, imu_data, can_data]):
return None
state = VehicleState()
state.timestamp = time.time()
# 填充GPS数据
if gps_data:
state.latitude = gps_data.get('latitude', 0.0)
state.longitude = gps_data.get('longitude', 0.0)
state.altitude = gps_data.get('altitude', 0.0)
state.speed_gps = gps_data.get('speed', 0.0)
state.heading = gps_data.get('heading', 0.0)
# 填充IMU数据
if imu_data:
state.acceleration_x = imu_data.get('acceleration_x', 0.0)
state.acceleration_y = imu_data.get('acceleration_y', 0.0)
state.acceleration_z = imu_data.get('acceleration_z', 0.0)
state.angular_velocity_x = imu_data.get('gyro_x', 0.0)
state.angular_velocity_y = imu_data.get('gyro_y', 0.0)
state.angular_velocity_z = imu_data.get('gyro_z', 0.0)
# 填充CAN数据
if can_data:
state.speed_wheel = can_data.get('speed_wheel', 0.0)
state.steering_angle = can_data.get('steering_angle', 0.0)
state.brake_pedal = can_data.get('brake_pedal', 0.0)
state.accelerator_pedal = can_data.get('accelerator_pedal', 0.0)
state.gear_position = can_data.get('gear_position', 0)
state.engine_rpm = can_data.get('engine_rpm', 0)
state.turn_signal_left = can_data.get('turn_signal_left', False)
state.turn_signal_right = can_data.get('turn_signal_right', False)
state.emergency_brake = can_data.get('emergency_brake', False)
# 速度融合（卡尔曼滤波简化版）
if gps_data and can_data:
# 使用轮速作为基础，GPS速度用于校正
gps_speed = gps_data.get('speed', 0.0)
wheel_speed = can_data.get('speed_wheel', 0.0)
state.speed_gps = 0.7 * wheel_speed + 0.3 * gps_speed
return state
def get_latest_state(self) -> Optional[VehicleState]:
"""
获取最新的车辆状态
Returns:
最新的车辆状态
"""
try:
return self.data_queue.get_nowait()
except queue.Empty:
return None
def get_data_statistics(self) -> Dict:
"""
获取数据统计信息
Returns:
统计信息字典
"""
return {
'queue_size': self.data_queue.qsize(),
'is_running': self.is_running,
'connected_sources': [
name for name, source in self.data_sources.items()
if source.is_connected()
],
'latest_gps': self.latest_gps_data,
'latest_imu': self.latest_imu_data,
'latest_can': self.latest_can_data
}
# 测试代码
if __name__ == "__main__":
# 创建数据采集管理器
manager = DataAcquisitionManager()
# 注册数据源
manager.register_data_source('gps', GPSDataSource())
manager.register_data_source('imu', IMUDataSource())
manager.register_data_source('can', CANBusDataSource())
# 连接所有数据源
if manager.connect_all():
print("所有数据源连接成功")
# 定义数据回调
def on_data_fused(state: VehicleState):
print(f"时间: {state.timestamp:.2f}, "
f"位置: ({state.latitude:.6f}, {state.longitude:.6f}), "
f"速度: {state.speed_gps:.2f} m/s, "
f"转向角: {state.steering_angle:.1f}°")
# 启动数据采集
manager.start_acquisition(on_data_fused)
# 运行10秒
time.sleep(10)
# 停止采集
manager.stop_acquisition()
manager.disconnect_all()
print("数据采集测试完成")
else:
print("数据源连接失败")
4.2 压线检测模块 (core/line_detection.py)
"""
压线检测模块
实现基于电子围栏的压线检测和轨迹偏离预警
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！