python如何让红绿灯看懂两侧车流智能调节红绿灯？

下面给你一套可落地的思路与示例代码：用 Python + CV 模型实时感知路口四个方向车流量，按需自适应分配绿灯时长，提升通勤效率。

一、整体方案（端到端）

感知层：路口相机（1 个俯视/斜俯视广角，或每个方向 1 个相机）→ RTSP 视频流

识别跟踪：YOLO 检测（car/bus/truck/motorcycle）+ ByteTrack 多目标跟踪

交通量指标：

车队长度/排队数：在“停止线前的队列区域 ROI”统计低速车辆数

到达率（可选）：在“上游计数线”统计单位时间过线数

PCU 权重：按车种折算等效小客车数（car=1, moto=0.3, bus=2, truck=2.5）

控制策略：自适应相位控制（最小绿/最大绿/黄灯/全红、按需求分配绿灯，支持gap-out）

执行层：模拟（UI 可视化/MQTT）或对接信号机（Modbus/RS485/PLC）

监控与安全：异常回落固定配时、日志与告警、夜间/恶劣天气策略

二、技术选型

模型：YOLOv8n/s（ultralytics），或 YOLO-World/RT-DETR（更强更慢）

跟踪：ByteTrack（supervision 库封装）

视频：OpenCV 读取 RTSP

加速硬件：Jetson Orin/Xavier、RTX 30/40 系列；CPU 也可但路口多流会吃力

仓库与数据：COCO 预训练可用，场景适配可微调（UA-DETRAC、BDD100K、VisDrone 等）

三、实现步骤1) 相机与标定

固定相机，尽量俯视减小遮挡；保存一帧清晰画面

标定停止线与“队列区域 Polygon（每个方向一块）”，可用鼠标点多边形或以真实尺寸做单应性标定（推荐）

可选：设置“上游计数线”，用于测到达率

2) 检测与跟踪

只保留车辆类（car/motorcycle/bus/truck）

跟踪获得稳定的 track_id 与位移，估计速度与方向

3) 车流量度量

队列判定：落在队列区域且速度低于阈值即视为排队

队列需求= Σ(车辆PCU权重)

相位（NS/EW）需求=对应两个方向的需求之和

可选：到达率叠加入需求，防止刚到达方向长期饥饿

4) 自适应控制（2 相位示例：南北同绿/东西同绿）

约束：最小绿（min_green）、最大绿（max_green）、黄灯（amber）、全红（all_red）

切换条件：

达到最小绿后，若对向需求显著更大或当前相位“gap-out”（一定时间无车/队列清空），则转黄→全红→对向

达最大绿强制切换

公平性：设置需求阈值与加权，防止某方向长时间不得绿

四、核心 Python 示例（YOLOv8 + ByteTrack + OpenCV + 简易自适应控制）依赖安装：pip install ultralytics supervision==0.18.0 opencv-python numpy

示例仅演示单画面（一个广角包含四个方向）。如果你是四路相机，读四个 RTSP 并在同一时钟下融合各自 ROI 即可。

import time
import cv2
import numpy as np
from ultralytics import YOLO
import supervision as sv

# 1) 配置区域（需要你根据画面标注）
# 每个方向的“队列区域”顶点（顺时针），像素坐标
QUEUE_ROIS = {
    "N": [(100, 100), (300, 100), (300, 260), (100, 260)],
    "S": [(100, 700), (300, 700), (300, 540), (100, 540)],
    "E": [(1000, 300), (1160, 300), (1160, 500), (1000, 500)],
    "W": [(20, 300), (180, 300), (180, 500), (20, 500)],
}
# 如需到达率，额外定义上游计数线（示例）
LINE_PAIRS = {
    "N": ((120, 260), (280, 260)),
    "S": ((120, 540), (280, 540)),
    "E": ((1000, 320), (1000, 480)),
    "W": ((180, 320), (180, 480)),
}

# PCU 权重
PCU = {2: 1.0, 3: 0.3, 5: 2.0, 7: 2.5}  # COCO: 2 car, 3 moto, 5 bus, 7 truck
INTEREST_CLASSES = set(PCU.keys())

# 速度阈值（像素/秒），建议完成单应性标定后改为 m/s 阈值如 0.5 m/s
QUEUE_SPEED_THRESH_PX = 12.0

# 2) 自适应控制器（两相位：NS/EW）
class AdaptiveController:
    def __init__(self, min_green=10, max_green=45, amber=3, all_red=1, gap_out=3, demand_bias=1.15):
        self.state = "NS_GREEN"  # NS_GREEN, NS_AMBER, ALL_RED_TO_EW, EW_GREEN, EW_AMBER, ALL_RED_TO_NS
        self.t_state = time.time()
        self.min_green = min_green
        self.max_green = max_green
        self.amber = amber
        self.all_red = all_red
        self.gap_out = gap_out
        self.demand_bias = demand_bias
        self.last_seen_vehicle = {"NS": time.time(), "EW": time.time()}

    def update_last_seen(self, phase, has_vehicle):
        if has_vehicle:
            self.last_seen_vehicle[phase] = time.time()

    def time_in_state(self):
        return time.time() - self.t_state

    def set_state(self, s):
        self.state = s
        self.t_state = time.time()

    def decide(self, demand_NS, demand_EW, has_NS, has_EW):
        # 更新“是否有车”观察，用于gap-out
        self.update_last_seen("NS", has_NS)
        self.update_last_seen("EW", has_EW)

        t = self.time_in_state()

        if self.state == "NS_GREEN":
            # 强制最小绿
            if t < self.min_green:
                return self.output("NS_GREEN")
            # 最大绿
            if t >= self.max_green:
                self.set_state("NS_AMBER")
        ...

点击查看剩余70%