Python比对两组传感器车牌数据 计算车速判定车辆超速

现有代码的核心问题 #

你当前写的代码存在几个硬伤，直接运行无法达到预期效果：

• 重复定义了两次speed函数，后定义的版本会直接覆盖前序版本，且两个版本都没有写返回逻辑，计算结果无法向外传递
• 重复读取了两次输入文件，平白多了一倍IO开销，面对海量数据时效率极低
• elapsed_time函数计算完时间差没有设置返回值，后续计算速度时根本取不到时间差结果
• 时间解析逻辑仅支持:分隔的时间格式，你样例中2号传感器的时间用.做分隔符（12.00.42），直接运行会触发报错
• 没有实现传感器数据拆分、同车牌记录匹配的核心逻辑，也没有脏数据处理分支，碰到字段缺失、仅被单个传感器拍到的车牌、时间顺序颠倒的记录会直接崩溃
• 打开文件没有使用上下文管理器，程序异常退出时会出现文件句柄泄漏问题

实现方案选型 #

你最初计划拆成两个列表再逐行比对的思路仅适合极小数据量场景：两个列表做同车牌匹配的时间复杂度是O(n²)，如果是几十万上百万条的海量记录，运行速度会非常慢。
更适配海量数据的方案是用字典做哈希匹配：

• 仅缓存1号传感器的采集记录，字典key为车牌号，value为转换成秒数的时间戳
• 遍历到2号传感器的记录时，直接通过key查询字典中是否存在同车牌的1号传感器记录，查到就计算时间差和行驶速度，算完直接把该key从字典中删除，不保留冗余数据
• 单条记录匹配的时间复杂度为O(1)，整体仅需遍历一遍文件即可输出结果，内存占用极低

修正后的完整代码 #

# 两个传感器之间的固定距离，单位：米
SENSOR_DISTANCE = 200

def parse_time_to_sec(time_str):
    # 统一替换点分隔符为冒号，兼容两种时间格式
    time_str = time_str.replace('.', ':')
    hours, minutes, seconds = map(int, time_str.split(':'))
    return hours * 3600 + minutes * 60 + seconds

def calc_actual_speed(elapsed_sec):
    # 速度换算：米/秒转km/h 乘以系数3.6
    if elapsed_sec <= 0:
        return 0
    return (SENSOR_DISTANCE / elapsed_sec) * 3.6

if __name__ == "__main__":
    # 用上下文管理器读取文件，自动释放文件句柄
    with open('traffic_input.txt', 'r', encoding='utf-8') as f:
        # 过滤空行、首尾空白字符
        lines = [line.strip() for line in f if line.strip()]
    
    # 解析第一行的限速值
    max_speed_limit = float(lines[0])
    # 存储1号传感器待匹配的车牌-时间映射
    sensor1_pending = {}
    over_speed_list = []

    for line in lines[1:]:
        parts = line.split()
        # 跳过字段数不对的脏数据
        if len(parts) != 3:
            continue
        sensor_id, time_str, plate = parts
        time_sec = parse_time_to_sec(time_str)
        
        if sensor_id == '1':
            sensor1_pending[plate] = time_sec
        elif sensor_id == '2':
            # 匹配到同车牌的1号传感器记录
            if plate in sensor1_pending:
                start_time = sensor1_pending.pop(plate)
                time_cost = time_sec - start_time
                # 过滤时间差异常的脏数据
                if time_cost <= 0:
                    continue
                actual_speed = calc_actual_speed(time_cost)
                if actual_speed > max_speed_limit:
                    over_speed_list.append(
                        f"车牌{plate} 实际时速{actual_speed:.1f}km/h，超出限速{max_speed_limit}km/h"
                    )
    
    # 输出结果，可根据需求修改为写入文件等逻辑
    print("超速车辆清单：")
    for item in over_speed_list:
        print(item)

关键逻辑说明 #

• 时间解析环节统一替换.为:，兼容两种分隔符的时间格式，不会因为数据格式不统一报错
• 所有读入行先做空白字符过滤，拆分后字段数不符合要求的脏数据直接跳过，不会中断整体运行
• 匹配完成的车牌直接从待匹配字典中弹出，内存中仅保留还未匹配到2号记录的车牌，数据量再大也不会出现内存占用过高的问题
• 自动过滤时间差小于等于0的异常记录（比如传感器数据顺序传反、时间戳错误的情况）
• 限速值直接从文件首行解析为数值类型，没有冗余的无效函数调用
• 速度计算、时间解析逻辑拆分为独立函数，后续调整距离参数、限速规则时仅需修改对应函数即可，不需要改动主流程

内容的提问来源于stack exchange，提问作者Hoi2022