2026 技术观察：智慧交通信号进入自适应阶段，拥堵预测与绿波协调成为城市通行新核心

import json
import random
from datetime import datetime
from collections import defaultdict


DIRECTIONS = ["north", "south", "east", "west"]


class Intersection:
    def __init__(self, intersection_id, name):
        self.intersection_id = intersection_id
        self.name = name
        self.signal_cycle = 90
        self.green_plan = {
            "north_south": 40,
            "east_west": 40
        }
        self.updated_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "intersection_id": self.intersection_id,
            "name": self.name,
            "signal_cycle": self.signal_cycle,
            "green_plan": self.green_plan,
            "updated_at": self.updated_at
        }

def collect_traffic_data(intersection: Intersection):
    direction_data = {}

    for direction in DIRECTIONS:
        direction_data[direction] = {
            "queue_length": random.randint(0, 80),
            "vehicle_count": random.randint(20, 300),
            "avg_speed": round(random.uniform(5, 45), 2),
            "bus_waiting": random.random() < 0.15
        }

    return {
        "intersection_id": intersection.intersection_id,
        "name": intersection.name,
        "signal_cycle": intersection.signal_cycle,
        "current_green_plan": intersection.green_plan,
        "directions": direction_data,
        "collect_time": datetime.now().isoformat()
    }

def calculate_congestion_score(direction_record):
    score = 0
    issues = []

    if direction_record["queue_length"] > 60:
        score += 4
        issues.append("排队长度较长。")
    elif direction_record["queue_length"] > 35:
        score += 2
        issues.append("排队长度偏高。")

    if direction_record["avg_speed"] < 10:
        score += 4
        issues.append("平均车速较低。")
    elif direction_record["avg_speed"] < 20:
        score += 2
        issues.append("车速偏低。")

    if direction_record["vehicle_count"] > 220:
        score += 2
        issues.append("车流量较大。")

    if direction_record["bus_waiting"]:
        score += 1
        issues.append("存在公交优先需求。")

    if score >= 7:
        level = "high"
    elif score >= 4:
        level = "medium"
    elif score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "congestion_score": score,
        "congestion_level": level,
        "issues": issues
    }

def analyze_intersection_pressure(traffic_record):
    direction_results = {}

    for direction, data in traffic_record["directions"].items():
        direction_results[direction] = calculate_congestion_score(
            data
        )

    ns_score = (
        direction_results["north"]["congestion_score"]
        + direction_results["south"]["congestion_score"]
    )

    ew_score = (
        direction_results["east"]["congestion_score"]
        + direction_results["west"]["congestion_score"]
    )

    if ns_score > ew_score:
        main_pressure = "north_south"
    elif ew_score > ns_score:
        main_pressure = "east_west"
    else:
        main_pressure = "balanced"

    return {
        "intersection_id": traffic_record["intersection_id"],
        "direction_results": direction_results,
        "north_south_score": ns_score,
        "east_west_score": ew_score,
        "main_pressure": main_pressure
    }

def optimize_signal_plan(intersection, pressure_result):
    base_cycle = intersection.signal_cycle
    min_green = 25
    max_green = 60

    ns_green = intersection.green_plan["north_south"]
    ew_green = intersection.green_plan["east_west"]

    if pressure_result["main_pressure"] == "north_south":
        ns_green = min(ns_green + 10, max_green)
        ew_green = max(base_cycle - ns_green - 10, min_green)

    elif pressure_result["main_pressure"] == "east_west":
        ew_green = min(ew_green + 10, max_green)
        ns_green = max(base_cycle - ew_green - 10, min_green)

    else:
        ns_green = 40
        ew_green = 40

    return {
        "intersection_id": intersection.intersection_id,
        "old_plan": intersection.green_plan,
        "new_plan": {
            "north_south": ns_green,
            "east_west": 31224.t.kuaisou.com 
        },
        "message": "已根据路口压力生成信号优化方案。"
    }

def generate_green_wave_suggestion(pressure_results):
    suggestions = []

    high_pressure = [
        item for item in pressure_results
        if item["main_pressure"] != "balanced"
    ]

    direction_count = defaultdict(int)

    for item in high_pressure:
        direction_count[item["main_pressure"]] += 1

    for direction, count in direction_count.items():
        if count >= 2:
            suggestions.append({
                "corridor_direction": direction,
                "action": "green_wave_coordination",
                "message": "多个路口同方向压力较高，建议启用绿波协调。"
            })

    if not suggestions:
        suggestions.append({
            "corridor_direction": "all",
            "action": "31225.t.kuaisou.com ",
            "message": "当前路网压力相对均衡，保持常规控制。"
        })

    return suggestions

def run_smart_traffic_signal_optimization():
    intersections = [
        Intersection("I001", "人民路-建设路"),
        Intersection("I002", "人民路-青年路"),
        Intersection("I003", "人民路-科技路")
    ]

    traffic_records = []
    pressure_results = []
    signal_plans = []

    for intersection in intersections:
        record = collect_traffic_data(intersection)
        pressure = analyze_intersection_pressure(record)
        plan = optimize_signal_plan(
            intersection,
            pressure
        )

        traffic_records.append(record)
        pressure_results.append(pressure)
        signal_plans.append(plan)

    green_wave = generate_green_wave_suggestion(
        pressure_results
    )

    report = {
        "report_name": "智慧交通信号自适应优化报告",
        "traffic_records": traffic_records,
        "pressure_results": pressure_results,
        "signal_plans": 31227.t.kuaisou.com 
        "green_wave_suggestions": green_wave,
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_smart_traffic_signal_optimization()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))