# Python红绿灯模拟实现详解 下面我将通过多线程编程和计算机视觉两个维度，详细展示Python实现红绿灯功能的具体代码。 ## 一、多线程红绿灯模拟实现 ### 基础红绿灯状态控制 ```python import threading import time import random class TrafficLight: def __init__(self): self.green_light = threading.Event() self.red_light = threading.Event() self.current_state = "RED" self.red_light.set() # 初始状态为红灯 def change_light(self): """红绿灯状态切换线程""" while True: if self.current_state == "RED": # 红灯变绿灯 self.red_light.clear() self.green_light.set() self.current_state = "GREEN" print("🚦 交通灯：红灯 → 绿灯") time.sleep(10) # 绿灯持续10秒 else: # 绿灯变红灯 self.green_light.clear() self.red_light.set() self.current_state = "RED" print("🚦 交通灯：绿灯 → 红灯") time.sleep(10) # 红灯持续10秒 def vehicle_behavior(vehicle_id, traffic_light): """车辆行为模拟线程""" print(f"🚗 车辆 {vehicle_id} 到达路口") if traffic_light.current_state == "RED": print(f"🚗 车辆 {vehicle_id} 遇到红灯，等待中...") traffic_light.red_light.wait() # 等待红灯信号 print(f"🚗 车辆 {vehicle_id} 遇到绿灯，通过路口") time.sleep(random.uniform(0.5, 2.0)) # 模拟通过时间 # 主程序执行 if __name__ == "__main__": traffic_light = TrafficLight() # 启动红绿灯控制线程 light_thread = threading.Thread(target=traffic_light.change_light) light_thread.daemon = True light_thread.start() # 模拟多辆车辆通过 vehicle_threads = [] for i in range(20): vehicle_thread = threading.Thread( target=vehicle_behavior, args=(i+1, traffic_light) ) vehicle_threads.append(vehicle_thread) vehicle_thread.start() time.sleep(random.uniform(0.5, 3)) # 随机时间间隔生成车辆 # 等待所有车辆线程完成 for thread in vehicle_threads: thread.join() ``` ### 增强版红绿灯系统 ```python import threading import time from datetime import datetime from collections import deque class AdvancedTrafficLight: def __init__(self): self.state_event = threading.Event() self.current_state = "RED" self.state_durations = { "RED": 15, # 红灯15秒 "GREEN": 12, # 绿灯12秒 "YELLOW": 3 # 黄灯3秒 } self.vehicle_queue = deque() self.vehicle_count = 0 self.state_event.set() def light_controller(self): """完整的红绿灯控制循环""" state_sequence = ["RED", "GREEN", "YELLOW"] current_index = 0 while True: current_state = state_sequence[current_index] self.current_state = current_state duration = self.state_durations[current_state] # 状态切换通知 timestamp = datetime.now().strftime("%H:%M:%S") state_symbols = { "RED": "🔴", "GREEN": "🟢", "YELLOW": "🟡" } print(f"{timestamp} {state_symbols[current_state]} 交通灯状态: {current_state} - 持续{duration}秒") # 如果是绿灯，通知等待车辆通过 if current_state == "GREEN": self.state_event.set() time.sleep(duration) # 切换到下一个状态 current_index = (current_index + 1) % len(state_sequence) if current_state == "GREEN": self.state_event.clear() def vehicle_crossing(self, vehicle_id): """车辆通过路口的行为""" arrival_time = datetime.now().strftime("%H:%M:%S") print(f"{arrival_time} 🚗 车辆 {vehicle_id} 到达路口") # 等待绿灯 if self.current_state != "GREEN": print(f" 车辆 {vehicle_id} 等待绿灯...") self.state_event.wait() crossing_time = datetime.now().strftime("%H:%M:%S") print(f"{crossing_time} 🚗 车辆 {vehicle_id} 通过路口") # 模拟通过时间 time.sleep(random.uniform(1, 2)) # 运行示例 def run_traffic_simulation(): traffic_system = AdvancedTrafficLight() # 启动红绿灯控制 controller_thread = threading.Thread(target=traffic_system.light_controller) controller_thread.daemon = True controller_thread.start() # 模拟车辆生成 def generate_vehicles(): for i in range(30): vehicle_thread = threading.Thread( target=traffic_system.vehicle_crossing, args=(i+1,) ) vehicle_thread.start() time.sleep(random.uniform(1, 4)) vehicle_generator = threading.Thread(target=generate_vehicles) vehicle_generator.start() vehicle_generator.join() if __name__ == "__main__": run_traffic_simulation() ``` ## 二、基于计算机视觉的红绿灯识别 ### OpenCV红绿灯颜色识别 ```python import cv2 import numpy as np from collections import deque class TrafficLightDetector: def __init__(self): # 定义颜色范围 (HSV格式) self.color_ranges = { 'red': [ (0, 120, 70), (10, 255, 255), # 红色范围1 (170, 120, 70), (180, 255, 255) # 红色范围2 ], 'yellow': [(20, 100, 100), (30, 255, 255)], 'green': [(40, 50, 50), (90, 255, 255)] } self.state_history = deque(maxlen=10) # 状态历史记录 def detect_traffic_light_color(self, frame): """检测红绿灯颜色""" # 转换为HSV颜色空间 hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) # 定义感兴趣区域 (假设红绿灯在图像上方) height, width = frame.shape[:2] roi = hsv[0:height//2, width//4:3*width//4] max_area = 0 detected_color = "UNKNOWN" for color, ranges in self.color_ranges.items(): if color == 'red': # 红色有两个范围 mask1 = cv2.inRange(roi, ranges[0], ranges[1]) mask2 = cv2.inRange(roi, ranges[2], ranges[3]) mask = mask1 + mask2 else: mask = cv2.inRange(roi, ranges[0], ranges[1]) # 形态学操作去除噪声 kernel = np.ones((5,5), np.uint8) mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel) mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel) # 计算颜色区域面积 area = cv2.countNonZero(mask) if area > max_area and area > 100: # 面积阈值 max_area = area detected_color = color.upper() return detected_color def process_video(self, video_path=0): """处理视频流进行红绿灯识别""" cap = cv2.VideoCapture(video_path) while True: ret, frame = cap.read() if not ret: break # 检测红绿灯颜色 color = self.detect_traffic_light_color(frame) self.state_history.append(color) # 基于历史记录确定稳定状态 stable_color = self._get_stable_state() # 在图像上显示结果 cv2.putText(frame, f"Traffic Light: {stable_color}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2) cv2.imshow('Traffic Light Detection', frame) if cv2.waitKey(1) & 0xFF == ord('q'): break cap.release() cv2.destroyAllWindows() def _get_stable_state(self): """根据历史记录确定稳定的红绿灯状态""" if len(self.state_history) < 5: return "UNKNOWN" # 统计最近状态 from collections import Counter most_common = Counter(self.state_history).most_common(1) return most_common[0][0] if most_common[0][1] >= 5 else "UNKNOWN" # 使用示例 if __name__ == "__main__": detector = TrafficLightDetector() # 从摄像头检测 # detector.process_video(0) # 从视频文件检测 # detector.process_video("traffic_light_video.mp4") ``` ### 基于YOLOv5的红绿灯检测 ```python import torch import cv2 import numpy as np class YOLOv5TrafficLightDetector: def __init__(self, model_path='yolov5s.pt'): """初始化YOLOv5模型""" self.model = torch.hub.load('ultralytics/yolov5', 'custom', path=model_path) self.model.conf = 0.5 # 置信度阈值 # 红绿灯类别映射 self.class_names = { 0: 'Red', 1: 'Green', 2: 'Yellow' } def detect_traffic_lights(self, image_path): """检测图像中的红绿灯""" # 进行推理 results = self.model(image_path) # 解析结果 detections = [] for *xyxy, conf, cls in results.xyxy[0]: if int(cls) in self.class_names: detection = { 'bbox': [int(coord) for coord in xyxy], 'confidence': float(conf), 'class': self.class_names[int(cls)], 'class_id': int(cls) } detections.append(detection) return detections, results def visualize_detection(self, image_path, save_path=None): """可视化检测结果""" detections, results = self.detect_traffic_lights(image_path) # 显示结果 results.show() if save_path: results.save(save_dir=save_path) return detections # 使用示例 def demo_yolov5_detection(): detector = YOLOv5TrafficLightDetector() # 检测单张图片 image_path = "traffic_light_image.jpg" detections = detector.visualize_detection(image_path) print("检测到的红绿灯:") for detection in detections: print(f"- {detection['class']}: 置信度 {detection['confidence']:.2f}") return detections if __name__ == "__main__": demo_yolov5_detection() ``` ## 三、两种实现方式的对比 | 特性 | 多线程模拟 | 计算机视觉识别 | |------|------------|----------------| | **应用场景** | 交通仿真、教学演示 | 自动驾驶、智能交通系统 | | **实现复杂度** | 中等，主要涉及线程同步 | 较高，需要图像处理知识 | | **实时性** | 高，纯代码执行 | 中等，依赖硬件性能 | | **准确性** | 100%准确（预设逻辑） | 依赖模型训练质量和环境条件 | | **扩展性** | 容易添加新功能 | 需要重新训练模型 | | **资源需求** | 低，仅需Python环境 | 高，需要GPU加速 | ## 四、实际应用建议 ### 1. 教学和演示场景 推荐使用多线程模拟实现，代码清晰易懂，能够很好地展示红绿灯的工作原理和线程同步概念 [ref_1]。 ### 2. 实际工程项目 对于智能交通系统或自动驾驶应用，建议采用基于YOLOv5的检测方案，准确性更高且更适合真实环境 [ref_2]。 ### 3. 性能优化技巧 - 多线程版本：合理设置线程数量，避免线程过多导致的性能下降 - 视觉识别版本：使用GPU加速推理过程，优化图像预处理流程 以上代码提供了从基础到高级的完整红绿灯实现方案，可以根据具体需求选择合适的实现方式。多线程版本适合理解概念和快速原型开发，而计算机视觉版本则更适合实际应用部署。