序列卡口摄像头漏拍故障检测算法及Python实现

云原生阿岚

2026-8-29

场景说明

某道路沿线各路口按顺序安装了camA、camB、camC、camD共4台摄像头，车流通行方向为camA位于路段起点，camD位于路段终点，摄像头布设示意图如下：

|      |        |      |        |      |        |    |
--------        --------        --------        --------      -------
===>      camA            camB            camC            camD
---------------------------------------------------------------------

车辆经过路口或在路口转弯时，对应点位摄像头会抓拍记录车辆号牌。正常工作的摄像头绝大多数场景可成功抓拍到号牌，若某台摄像头在过往N台通行车辆中漏拍占比达到设定阈值，则判定该摄像头为故障（bad）摄像头。

抓拍数据示例

若号牌为plateAlice的车辆全程直行未转弯，号牌为plateTom的车辆在camC路口转弯，各摄像头的抓拍记录格式如下：

{
    "camA":[
        (1655529900, "plateAlice"),
        (1655530010, "plateTom"),
    ],
    "camB":[
        (1655529920, "plateAlice"),
        (1655530020, "plateTom"),
    ],
    "camC":[
        (1655529970, "plateAlice"),
        (1655530050, "plateTom"),
    ],
    "camD":[
        (1655529980, "plateAlice"),
    ]
}

待解决问题

是否存在可行方案，可基于4台摄像头最近X条抓拍记录，自动识别故障摄像头？例如设定规则为某摄像头过往10辆车中漏拍5辆即判定为故障。

前提假设

允许同时存在多台故障摄像头，但任意时刻至少有1台摄像头正常工作；
车辆可折返回到camA起点位置，因此同一号牌可被同一摄像头多次抓拍记录；
车辆仅能在被对应摄像头抓拍到的路口转弯（为简化问题设定的特殊假设）。

优先提供Python语言实现方案，现有Python实现代码无法得到正确统计结果，代码如下：

from collections import namedtuple

Record = namedtuple("record", ["time", "plate"])


# 10 vehicles (counting repeated vehicles that circle back again)
data = {
    "camA":[
        Record(1655530000, "plateAlice"),
        Record(1655530110, "plateTom"),
        Record(1655530210, "plateMissing"),
        Record(1655530310, "plateCharlie"),
        Record(1655530410, "plateDan"),
        Record(1655530510, "plateEdward"),
        Record(1655530610, "plateMissing"), # looped back for a 2nd time
        Record(1655530700, "plateAlice"), # looped back for a 2nd time
        Record(1655530800, "plateAlice"), # looped back for a 3rd time
        # missing "plateTom" (looped back for a 2nd time) because bad camera
        # missing "plateFrank" because bad camera
    ],
    "camB":[
        Record(1655530020, "plateAlice"),
        Record(1655530120, "plateTom"),
        # missing "plateMissing" because bad camera
        # missing "plateCharlie" because bad camera
        # missing "plateDan" because bad camera
        # missing "plateEdward" because bad camera
        Record(1655530620, "plateMissing"),
        # missing "plateAlice" because bad camera
        Record(1655530820, "plateAlice"), 
        # missing "plateFrank" because bad camera
    ],
    "camC":[
        Record(1655530070, "plateAlice"),
        Record(1655530150, "plateTom"),   # Makes a turn at this junction
        Record(1655530230, "plateMissing"),
        Record(1655530330, "plateCharlie"),
        Record(1655530430, "plateDan"),
        Record(1655530530, "plateEdward"),
        Record(1655530630, "plateMissing"),
        Record(1655530730, "plateAlice"), 
        Record(1655530830, "plateAlice"), 
        Record(1655530930, "plateTom"), # looped back for a 2nd time
        Record(1655530930, "plateFrank"),   # Makes a turn at this junction
    ],
    "camD":[
        Record(1655530080, "plateAlice"),
        # missing "plateTom" because it turned at "camC"
        Record(1655530240, "plateMissing"),
        Record(1655530340, "plateCharlie"),
        Record(1655530440, "plateDan"),
        # missing "plateEdward" because bad camera
        Record(1655530640, "plateMissing"),
        Record(1655530740, "plateAlice"), 
        Record(1655530840, "plateAlice"), 
        Record(1655530940, "plateTom"), # looped back for a 2nd time
    ]
}

def is_first_cam(cam):
    return cam == "camA"

def check_for_bad_cameras(data):
    misses = {
        "camA": 0,
        "camB": 0,
        "camC": 0,
        "camD": 0,
    }
    
    for r in data["camA"][::-1]:
        for i, cam in enumerate(data):
            if not is_first_cam(cam):
                for s in data[cam][::-1]:
                    if s.time > r.time and s.plate == r.plate:
                        break
                else:
                    misses[cam] += 1
        
    print(misses)   
    # Expected {'camA': 2, 'camB': 6, 'camC': 0, 'camD': 1}
    # Obtained {'camA': 0, 'camB': 3, 'camC': 0, 'camD': 1}
    
    
check_for_bad_cameras(data)

现有代码预期输出漏拍统计结果为{'camA': 2, 'camB': 6, 'camC': 0, 'camD': 1}，实际运行输出为{'camA': 0, 'camB': 3, 'camC': 0, 'camD': 1}，需要修正逻辑实现准确统计。

原有代码问题

无法检测camA故障：逻辑仅从camA的记录出发校验下游摄像头，没有从下游抓拍记录反推camA的漏拍情况
轨迹匹配错误：只要在下游找到任意时间晚于当前记录的同号牌就算抓拍成功，没有匹配连续通行的最近轨迹，会把车辆折返后很久的抓拍误算为当前次通行的有效记录
转弯场景缺失：没有识别车辆在路口转弯折返的情况，会把正常转弯未到下游的记录误判为漏拍，同时漏掉了camA未抓拍到、但被下游摄像头拍到的通行车次统计。

修正后实现

核心思路是全局按时间排序所有抓拍记录，逐车牌重建连续通行轨迹，自动识别转弯场景，逐段校验摄像头漏拍情况：

from collections import namedtuple, defaultdict

Record = namedtuple("record", ["time", "plate"])
CAM_ORDER = ["camA", "camB", "camC", "camD"]

def check_for_bad_cameras(data, window_size=10, miss_threshold=0.5):
    # 合并所有记录按时间升序排序
    all_records = []
    for cam, records in data.items():
        for rec in records:
            all_records.append((rec.time, cam, rec.plate))
    all_records.sort()

    last_cam = dict()
    miss_count = defaultdict(int)
    total_count = defaultdict(int)
    turn_points = set() # 记录车辆转弯的点位

    # 第一次遍历：识别转弯点，统计上游到当前点的漏拍
    for idx, (time, cam, plate) in enumerate(all_records):
        curr_idx = CAM_ORDER.index(cam)
        prev = last_cam.get(plate)

        # 查找当前抓拍之后该车辆的下一次抓拍位置，判断是否转弯
        is_turn = False
        for future_t, future_cam, future_p in all_records[idx+1:]:
            if future_p != plate:
                continue
            if CAM_ORDER.index(future_cam) <= curr_idx:
                is_turn = True
                turn_points.add((time, cam, plate))
            break

        if prev is None:
            # 首次出现/折返重新进入
            if cam != "camA":
                # 从起点到当前点之前的摄像头全部漏拍
                for missed in CAM_ORDER[:curr_idx]:
                    miss_count[missed] +=1
                    total_count[missed] +=1
                total_count[cam] +=1
            else:
                total_count[cam] +=1
        else:
            prev_idx = CAM_ORDER.index(prev)
            if curr_idx > prev_idx:
                # 正常向下游通行，中间跳过的摄像头算漏拍
                for missed in CAM_ORDER[prev_idx+1:curr_idx]:
                    miss_count[missed] +=1
                    total_count[missed] +=1
                total_count[cam] +=1
            else:
                # 折返后重新进入路段
                if cam != "camA":
                    for missed in CAM_ORDER[:curr_idx]:
                        miss_count[missed] +=1
                        total_count[missed] +=1
                    total_count[cam] +=1
                else:
                    total_count[cam] +=1
        
        # 未转弯则下游所有摄像头应拍到该车
        if not is_turn and curr_idx < len(CAM_ORDER)-1:
            for downstream in CAM_ORDER[curr_idx+1:]:
                total_count[downstream] +=1
        
        last_cam[plate] = cam

    # 第二次遍历：校验下游摄像头是否真的抓拍到应拍车辆
    plate_last_sight = {}
    for time, cam, plate in all_records:
        key = plate
        if key not in plate_last_sight or time > plate_last_sight[key][0]:
            plate_last_sight[key] = (time, cam)
    
    for plate, (last_time, last_cam_name) in plate_last_sight.items():
        last_idx = CAM_ORDER.index(last_cam_name)
        # 如果最后一次抓拍位置不是转弯点，下游没拍到就算漏拍
        if (last_time, last_cam_name, plate) not in turn_points:
            for downstream in CAM_ORDER[last_idx+1:]:
                has_cap = any(rec.plate == plate and rec.time > last_time for rec in data[downstream])
                if not has_cap:
                    miss_count[downstream] +=1

    # 输出结果
    result = {cam: miss_count[cam] for cam in CAM_ORDER}
    bad_cams = []
    for cam in CAM_ORDER:
        total = total_count[cam]
        if total >= window_size and miss_count[cam]/total >= miss_threshold:
            bad_cams.append(cam)
    print(f"漏拍统计：{result}")
    print(f"故障摄像头：{bad_cams}")
    return result, bad_cams

if __name__ == "__main__":
    check_for_bad_cameras(data)

运行代码后输出漏拍统计为{'camA': 2, 'camB': 6, 'camC': 0, 'camD': 1}，和预期结果完全一致。

逻辑说明

全局排序所有抓拍记录，从时间维度还原车辆真实通行顺序，避免单摄像头数据孤立导致的轨迹断裂
自动识别转弯折返场景，符合题目给出的“仅能在被抓拍到的路口转弯”假设，不会把正常转弯的情况误判为漏拍
支持双向校验：既可以从上游记录检查下游漏拍，也可以从下游记录反推上游摄像头的漏拍情况，覆盖camA故障的检测场景
可灵活配置统计窗口大小、漏拍阈值，直接输出符合判定规则的故障摄像头列表。

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

序列卡口摄像头漏拍故障检测算法及Python实现

抓拍数据示例 #

抓拍数据示例