PowLu / 第一版主程序.py

import cv2 import threading import numpy as np import time import os import serial if 'DISPLAY' not in os.environ: os.environ['DISPLAY'] = ':0' print(f"Set DISPLAY to: {os.environ['DISPLAY']}") import pyautogui # ---------- 1. 固定参数区 ---------- MODEL_PATH = './yolov5s-640-640.rknn' # 模型路径 TARGET = 'rk3588' # 目标设备 IMG_SHOW = True # 是否弹窗显示结果 IMG_SAVE = False # 是否保存结果图片 COCO_MAP_TEST = False # 是否跑COCO mAP测试 FULLSCREEN = True # 是否全屏显示 SOURCES_FILE = './sources.json' # 视频源配置文件 ANCHORS_FILE = './model/anchors_yolov5.txt' # anchor文件 # 优化参数 DETECTION_INTERVAL = 5 # 检测间隔（每5帧检测一次） DISPLAY_REFRESH_RATE = 30 # 显示刷新率(Hz) # 屏幕尺寸 SCREEN_WIDTH = 1024 SCREEN_HEIGHT = 600 # 显示区域尺寸 VIDEO_WIDTH = int(SCREEN_WIDTH * 0.75) # 左侧四分之三显示视频 RADIO_WIDTH = SCREEN_WIDTH - VIDEO_WIDTH # 右侧四分之一显示雷达 # ---------- 2. 雷达触发参数 ---------- ALARM_DISTANCE = 1.5 # 报警距离（米） SWITCH_INTERVAL = 3.0 # 切换间隔（秒） # ---------- 3. 其他全局常量 ---------- OBJ_THRESH = 0.6 NMS_THRESH = 0.6 IMG_SIZE = (640, 640) # (width, height) CLASSES = ("person", "bicycle", "car", "motorbike ", "aeroplane ", "bus ", "train", "truck ", "boat", "traffic light") # -------------------------- # RTSP流配置部分 - 简化版本 # -------------------------- RTSP_URLS = [ "rtsp://admin:Admin888@192.168.112.200:554/streaming/channels/102", "rtsp://admin:Admin888@192.168.112.201:554/streaming/channels/102", "rtsp://admin:Admin888@192.168.112.202:554/streaming/channels/102", "rtsp://admin:Admin888@192.168.112.203:554/streaming/channels/102" ] # 全局变量 frames = [None] * 4 locks = [threading.Lock() for _ in range(4)] stop_threads = False combined_frame = None combined_lock = threading.Lock() # 显示模式控制 display_mode = 0 # 0:四分屏全屏, 1-4:单画面分屏 last_switch_time = time.time() current_alarm_index = 0 # 雷达数据相关 radar_distances = [999.0] * 4 # 存储4路雷达距离（米） radar_alarm_status = [False] * 4 # 雷达报警状态 radar_lock = threading.Lock() # 雷达图片相关 radio_images = {} # 存储4个雷达图片 current_radio_image = None # 简化的流状态管理 stream_status = [False] * 4 # 每路流的连接状态 stream_last_frame_time = [0] * 4 # 每路流最后收到帧的时间 # ---------- 简化的RTSP流处理 ---------- def read_rtsp_stream(index, rtsp_url): """简化的RTSP流读取函数""" global stop_threads cap = None reconnect_count = 0 max_reconnect = 20 print(f"[Stream {index}] 启动RTSP流: {rtsp_url}") while not stop_threads and reconnect_count < max_reconnect: try: # 释放之前的连接 if cap is not None: cap.release() cap = None # 创建新的连接 print(f"[Stream {index}] 尝试连接...") cap = cv2.VideoCapture(rtsp_url) # 设置连接参数 cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) cap.set(cv2.CAP_PROP_FPS, 15) cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*'H264')) cap.set(cv2.CAP_PROP_OPEN_TIMEOUT_MSEC, 5000) cap.set(cv2.CAP_PROP_READ_TIMEOUT_MSEC, 3000) if not cap.isOpened(): print(f"[Stream {index}] 连接失败") reconnect_count += 1 time.sleep(3) continue print(f"[Stream {index}] ✓ 连接成功") stream_status[index] = True reconnect_count = 0 # 持续读取帧 while not stop_threads and cap.isOpened(): ret, frame = cap.read() if ret: with locks[index]: frames[index] = frame.copy() stream_last_frame_time[index] = time.time() stream_status[index] = True else: print(f"[Stream {index}] ✗ 读取帧失败") stream_status[index] = False break # 控制读取频率 time.sleep(0.03) except Exception as e: print(f"[Stream {index}] 异常: {e}") stream_status[index] = False # 连接断开，准备重连 if cap is not None: cap.release() cap = None reconnect_count += 1 print(f"[Stream {index}] 等待重连 ({reconnect_count}/{max_reconnect})...") time.sleep(3) print(f"[Stream {index}] 线程退出") # ---------- 简化的画面合成函数 ---------- def combine_frames(): global combined_frame, stop_threads while not stop_threads: frame_list = [] current_time = time.time() for i in range(4): with locks[i]: # 检查流是否超时（5秒无数据认为超时） if current_time - stream_last_frame_time[i] > 5.0: stream_status[i] = False frame_list.append(None) elif frames[i] is not None: frame_list.append(frames[i]) else: frame_list.append(None) # 检查是否有有效帧 valid_frames = [frame for frame in frame_list if frame is not None] if len(valid_frames) == 0: time.sleep(0.1) continue # 获取参考尺寸（使用第一个有效帧） first_valid_frame = next(frame for frame in frame_list if frame is not None) h, w, _ = first_valid_frame.shape # 创建合成画面 combined = np.zeros((h * 2, w * 2, 3), dtype=np.uint8) positions = [ (0, 0), # 左上 - Stream 0 (0, w), # 右上 - Stream 1 (h, 0), # 左下 - Stream 2 (h, w) # 右下 - Stream 3 ] for i, (y, x) in enumerate(positions): if frame_list[i] is not None and stream_status[i]: resized_frame = cv2.resize(frame_list[i], (w, h)) combined[y:y + h, x:x + w] = resized_frame # 添加状态指示器 status_color = (0, 255, 0) # 绿色 - 在线 status_text = f"Cam{i + 1} ✓" else: # 显示无信号画面 no_signal = create_single_no_signal(i + 1, w, h) combined[y:y + h, x:x + w] = no_signal status_color = (0, 0, 255) # 红色 - 离线 status_text = f"Cam{i + 1} ✗" # 在画面角落添加状态指示 cv2.putText(combined, status_text, (x + 10, y + 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, status_color, 2) with combined_lock: combined_frame = combined.copy() time.sleep(0.03) # ---------- 雷达串口读取类 ---------- class RadarUARTReader: def __init__(self, port: str = '/dev/ttyS0', baudrate: int = 115200): self.port = port self.baudrate = baudrate self.ser = None self.running = False self.receive_thread = None def start(self) -> bool: """启动雷达UART读取""" try: self.ser = serial.Serial( port=self.port, baudrate=self.baudrate, bytesize=8, parity='N', stopbits=1, timeout=0.1 ) self.running = True self.receive_thread = threading.Thread(target=self._receive_worker) self.receive_thread.daemon = True self.receive_thread.start() print(f"雷达数据读取器已启动: {self.port} @ {self.baudrate}bps") return True except Exception as e: print(f"雷达启动失败: {e}") return False def _parse_frame(self, frame_data: bytearray): """解析雷达数据帧""" if len(frame_data) != 16: return None # 验证帧头帧尾 if frame_data[0] != 0xA5 or frame_data[1] != 0x5A or frame_data[14] != 0x5A or frame_data[15] != 0xA5: return None radar_data_list = [] for i in range(4): start_idx = 2 + i * 3 radar_id = frame_data[start_idx] distance_high = frame_data[start_idx + 1] distance_low = frame_data[start_idx + 2] distance_mm = (distance_high << 8) | distance_low distance_m = distance_mm / 1000.0 radar_data = { 'radar_id': radar_id, 'distance_mm': distance_mm, 'distance_m': distance_m } radar_data_list.append(radar_data) return radar_data_list def _receive_worker(self): """接收数据工作线程""" state = 0 # 0:等待头, 1:接收数据 frame_buffer = bytearray() while self.running and self.ser and self.ser.is_open: try: bytes_to_read = min(self.ser.in_waiting, 1024) if bytes_to_read > 0: data = self.ser.read(bytes_to_read) for byte in data: if state == 0: if byte == 0xA5: frame_buffer = bytearray([byte]) state = 1 elif state == 1: frame_buffer.append(byte) if len(frame_buffer) == 16: # 完整帧接收完成 radar_data = self._parse_frame(frame_buffer) if radar_data: self._update_radar_distances(radar_data) state = 0 frame_buffer = bytearray() elif len(frame_buffer) > 16: state = 0 frame_buffer = bytearray() except Exception as e: print(f"雷达接收错误: {e}") time.sleep(0.01) def _update_radar_distances(self, radar_data_list): """更新雷达距离数据""" global radar_distances, radar_alarm_status with radar_lock: for data in radar_data_list: radar_id = data['radar_id'] if 1 <= radar_id <= 4: distance = data['distance_m'] radar_distances[radar_id - 1] = distance radar_alarm_status[radar_id - 1] = (distance <= ALARM_DISTANCE and distance > 0) # 打印报警信息 if radar_alarm_status[radar_id - 1]: print(f"🚨 雷达{radar_id}报警! 距离: {distance:.2f}m") def stop(self): """停止读取""" self.running = False if self.ser and self.ser.is_open: self.ser.close() print("雷达数据读取器已停止") # ---------- 核心函数 ---------- def filter_boxes(boxes, box_confidences, box_class_probs): box_confidences = box_confidences.reshape(-1) class_max_score = np.max(box_class_probs, axis=-1) classes = np.argmax(box_class_probs, axis=-1) person_mask = (classes == 0) _class_pos = np.where((class_max_score * box_confidences >= OBJ_THRESH) & person_mask) scores = (class_max_score * box_confidences)[_class_pos] boxes = boxes[_class_pos] classes = classes[_class_pos] return boxes, classes, scores def nms_boxes(boxes, scores): x = boxes[:, 0] y = boxes[:, 1] w = boxes[:, 2] - boxes[:, 0] h = boxes[:, 3] - boxes[:, 1] areas = w * h order = scores.argsort()[::-1] keep = [] while order.size > 0: i = order[0] keep.append(i) xx1 = np.maximum(x[i], x[order[1:]]) yy1 = np.maximum(y[i], y[order[1:]]) xx2 = np.minimum(x[i] + w[i], x[order[1:]] + w[order[1:]]) yy2 = np.minimum(y[i] + h[i], y[order[1:]] + h[order[1:]]) w1 = np.maximum(0.0, xx2 - xx1 + 0.00001) h1 = np.maximum(0.0, yy2 - yy1 + 0.00001) inter = w1 * h1 ovr = inter / (areas[i] + areas[order[1:]] - inter) inds = np.where(ovr <= NMS_THRESH)[0] order = order[inds + 1] keep = np.array(keep) return keep def box_process(position, anchors): grid_h, grid_w = position.shape[2:4] col, row = np.meshgrid(np.arange(0, grid_w), np.arange(0, grid_h)) col = col.reshape(1, 1, grid_h, grid_w) row = row.reshape(1, 1, grid_h, grid_w) grid = np.concatenate((col, row), axis=1) stride = np.array([IMG_SIZE[1] // grid_h, IMG_SIZE[0] // grid_w]).reshape(1, 2, 1, 1) col = col.repeat(len(anchors), axis=0) row = row.repeat(len(anchors), axis=0) anchors = np.array(anchors) anchors = anchors.reshape(*anchors.shape, 1, 1) box_xy = position[:, :2, :, :] * 2 - 0.5 box_wh = pow(position[:, 2:4, :, :] * 2, 2) * anchors box_xy += grid box_xy *= stride box = np.concatenate((box_xy, box_wh), axis=1) xyxy = np.copy(box) xyxy[:, 0, :, :] = box[:, 0, :, :] - box[:, 2, :, :] / 2 xyxy[:, 1, :, :] = box[:, 1, :, :] - box[:, 3, :, :] / 2 xyxy[:, 2, :, :] = box[:, 0, :, :] + box[:, 2, :, :] / 2 xyxy[:, 3, :, :] = box[:, 1, :, :] + box[:, 3, :, :] / 2 return xyxy def post_process(input_data, anchors): boxes, scores, classes_conf = [], [], [] input_data = [_in.reshape([len(anchors[0]), -1] + list(_in.shape[-2:])) for _in in input_data] for i in range(len(input_data)): boxes.append(box_process(input_data[i][:, :4, :, :], anchors[i])) scores.append(input_data[i][:, 4:5, :, :]) classes_conf.append(input_data[i][:, 5:, :, :]) def sp_flatten(_in): ch = _in.shape[1] _in = _in.transpose(0, 2, 3, 1) return _in.reshape(-1, ch) boxes = [sp_flatten(_v) for _v in boxes] classes_conf = [sp_flatten(_v) for _v in classes_conf] scores = [sp_flatten(_v) for _v in scores] boxes = np.concatenate(boxes) classes_conf = np.concatenate(classes_conf) scores = np.concatenate(scores) boxes, classes, scores = filter_boxes(boxes, scores, classes_conf) nboxes, nclasses, nscores = [], [], [] for c in set(classes): inds = np.where(classes == c) b = boxes[inds] c = classes[inds] s = scores[inds] keep = nms_boxes(b, s) if len(keep) != 0: nboxes.append(b[keep]) nclasses.append(c[keep]) nscores.append(s[keep]) if not nclasses and not nscores: return None, None, None boxes = np.concatenate(nboxes) classes = np.concatenate(nclasses) scores = np.concatenate(nscores) return boxes, classes, scores def draw(image, boxes, scores, classes): """正确的人体检测框绘制函数""" if boxes is None or len(boxes) == 0: return image box_color = (0, 0, 255) box_thickness = 3 text_color = (0, 0, 255) text_thickness = 2 for box, score, cl in zip(boxes, scores, classes): x1, y1, x2, y2 = [int(_b) for _b in box] x1 = max(0, min(x1, image.shape[1] - 1)) y1 = max(0, min(y1, image.shape[0] - 1)) x2 = max(0, min(x2, image.shape[1] - 1)) y2 = max(0, min(y2, image.shape[0] - 1)) cv2.rectangle(image, (x1, y1), (x2, y2), box_color, box_thickness) text = f'{CLASSES[cl]} {score:.2f}' cv2.putText(image, text, (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, text_color, text_thickness) return image def setup_model(model_path, device_id=None): """为每个NPU核心创建独立模型实例""" if model_path.endswith('.pt') or model_path.endswith('.torchscript'): platform = 'pytorch' from py_utils.pytorch_executor import Torch_model_container model = Torch_model_container(model_path) elif model_path.endswith('.rknn'): platform = 'rknn' from py_utils.rknn_executor import RKNN_model_container model = RKNN_model_container(model_path, TARGET, device_id) elif model_path.endswith('onnx'): platform = 'onnx' from py_utils.onnx_executor import ONNX_model_container model = ONNX_model_container(model_path) else: raise RuntimeError(f"{model_path} is not rknn/pytorch/onnx model") print(f'Model {model_path} initialized for device {device_id}') return model, platform # ---------- COCO_test_helper类 ---------- class COCO_test_helper: def __init__(self, enable_letter_box=True): self.enable_letter_box = enable_letter_box self.image_ids = [] self.category_ids = [] self.bboxes = [] self.scores = [] self.pad_top = 0 self.pad_left = 0 self.scale_ratio = 1.0 self.original_shape = (0, 0) self.new_shape = (0, 0) def letter_box(self, im, new_shape, pad_color=(0, 0, 0)): shape = im.shape[:2] if isinstance(new_shape, int): new_shape = (new_shape, new_shape) r = min(new_shape[0] / shape[0], new_shape[1] / shape[1]) new_unpad = int(round(shape[1] * r)), int(round(shape[0] * r)) dw, dh = new_shape[1] - new_unpad[0], new_shape[0] - new_unpad[1] dw /= 2 dh /= 2 if shape[::-1] != new_unpad: im = cv2.resize(im, new_unpad, interpolation=cv2.INTER_LINEAR) top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1)) left, right = int(round(dw - 0.1)), int(round(dw + 0.1)) self.pad_top = top self.pad_left = left self.scale_ratio = r self.original_shape = shape self.new_shape = new_shape im = cv2.copyMakeBorder(im, top, bottom, left, right, cv2.BORDER_CONSTANT, value=pad_color) return im def get_real_box(self, boxes): if boxes is None or len(boxes) == 0: return boxes real_boxes = [] for box in boxes: x1, y1, x2, y2 = box x1 = (x1 - self.pad_left) / self.scale_ratio y1 = (y1 - self.pad_top) / self.scale_ratio x2 = (x2 - self.pad_left) / self.scale_ratio y2 = (y2 - self.pad_top) / self.scale_ratio x1 = max(0, min(x1, self.original_shape[1])) y1 = max(0, min(y1, self.original_shape[0])) x2 = max(0, min(x2, self.original_shape[1])) y2 = max(0, min(y2, self.original_shape[0])) real_boxes.append([x1, y1, x2, y2]) return np.array(real_boxes) # ---------- 加载雷达图片 ---------- def load_radio_images(): """加载1-4号雷达图片""" global radio_images for i in range(1, 5): try: img_path = f'{i}.png' radio_img = cv2.imread(img_path) if radio_img is None: print(f"Warning: Cannot load {img_path}, creating default radar image") radio_img = create_default_radio(i) else: print(f"✓ Loaded radar image: {img_path}") radio_images[i] = cv2.resize(radio_img, (RADIO_WIDTH, SCREEN_HEIGHT)) except Exception as e: print(f"Failed to load radar image {i}.png: {e}") radio_images[i] = create_default_radio(i) return len(radio_images) > 0 def create_default_radio(radar_id): """创建默认雷达图片""" radio_img = np.zeros((SCREEN_HEIGHT, RADIO_WIDTH, 3), dtype=np.uint8) radio_img[:] = (20, 20, 20) center_x = RADIO_WIDTH // 2 center_y = SCREEN_HEIGHT // 2 radius = min(RADIO_WIDTH, SCREEN_HEIGHT) // 3 # 绘制雷达圆圈 for r in range(radius, 0, -radius // 4): color = (0, 255, 0) cv2.circle(radio_img, (center_x, center_y), r, color, 2) # 绘制雷达线 for angle in range(0, 360, 30): rad = np.deg2rad(angle) end_x = int(center_x + radius * np.cos(rad)) end_y = int(center_y + radius * np.sin(rad)) cv2.line(radio_img, (center_x, center_y), (end_x, end_y), (0, 255, 0), 1) # 显示雷达编号 title = f"Radar {radar_id}" title_size = cv2.getTextSize(title, cv2.FONT_HERSHEY_SIMPLEX, 0.8, 2)[0] title_x = (RADIO_WIDTH - title_size[0]) // 2 cv2.putText(radio_img, title, (title_x, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2) return radio_img # ---------- 雷达触发显示逻辑 ---------- def update_display_mode_by_radar(): """根据雷达数据更新显示模式""" global display_mode, last_switch_time, current_alarm_index, current_radio_image current_time = time.time() with radar_lock: # 获取报警的雷达列表 alarm_radars = [i for i in range(4) if radar_alarm_status[i]] if not alarm_radars: # 没有报警，显示四分屏 if display_mode != 0: display_mode = 0 current_radio_image = None print("✓ No alarm, switching to Quad View") return # 有报警，检查是否需要切换 if current_time - last_switch_time >= SWITCH_INTERVAL: if len(alarm_radars) == 1: # 单路报警，直接显示 new_mode = alarm_radars[0] + 1 if display_mode != new_mode: display_mode = new_mode current_radio_image = radio_images.get(new_mode) print(f"✓ Single alarm: Switching to Camera {new_mode}") else: # 多路报警，循环显示 current_alarm_index = (current_alarm_index + 1) % len(alarm_radars) new_mode = alarm_radars[current_alarm_index] + 1 display_mode = new_mode current_radio_image = radio_images.get(new_mode) print(f"✓ Multiple alarms: Cycling to Camera {new_mode}") last_switch_time = current_time # ---------- 人体检测处理 ---------- def process_frame_for_detection(frame, model, platform, co_helper, anchors): if frame is None: return None, None, None original_h, original_w = frame.shape[:2] pad_color = (0, 0, 0) img = co_helper.letter_box(im=frame.copy(), new_shape=(IMG_SIZE[1], IMG_SIZE[0]), pad_color=pad_color) img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) if platform in ['pytorch', 'onnx']: input_data = img.transpose(2, 0, 1) input_data = input_data.reshape(1, *input_data.shape).astype(np.float32) / 255. else: input_data = img outputs = model.run([np.expand_dims(input_data, 0)]) boxes, classes, scores = post_process(outputs, anchors) if boxes is not None: boxes = co_helper.get_real_box(boxes) return boxes, classes, scores # ---------- 窗口设置函数 ---------- def setup_borderless_fullscreen_window(): window_name = 'RTSP Surveillance System' cv2.setUseOptimized(True) cv2.namedWindow(window_name, cv2.WND_PROP_FULLSCREEN) cv2.resizeWindow(window_name, SCREEN_WIDTH, SCREEN_HEIGHT) cv2.moveWindow(window_name, 0, 0) cv2.setWindowProperty(window_name, cv2.WND_PROP_FULLSCREEN, cv2.WINDOW_FULLSCREEN) cv2.setWindowProperty(window_name, cv2.WND_PROP_TOPMOST, 1) print(f"✓ Borderless fullscreen window created: {SCREEN_WIDTH}x{SCREEN_HEIGHT}") print("✓ Window set to topmost and borderless") return window_name def hide_mouse(): try: pyautogui.moveTo(1024, 600) pyautogui.moveRel(0, 0) except Exception as e: print(f"Note: Could not hide mouse cursor: {e}") # ---------- 创建显示帧 ---------- def create_display_frame(display_mode, current_combined, detection_cache, frame_counter): """根据显示模式创建显示帧""" global current_radio_image, radar_distances if display_mode == 0: # 四分屏模式 - 全屏显示 if current_combined is None: display_frame = create_no_signal_frame(SCREEN_WIDTH, SCREEN_HEIGHT) else: display_frame = cv2.resize(current_combined, (SCREEN_WIDTH, SCREEN_HEIGHT)) if detection_cache['boxes'] is not None: original_h, original_w = current_combined.shape[:2] scale_x = SCREEN_WIDTH / original_w scale_y = SCREEN_HEIGHT / original_h scaled_boxes = [] for box in detection_cache['boxes']: x1, y1, x2, y2 = box scaled_boxes.append([ x1 * scale_x, y1 * scale_y, x2 * scale_x, y2 * scale_y ]) display_frame = draw(display_frame, np.array(scaled_boxes), detection_cache['scores'], detection_cache['classes']) # 显示雷达状态 with radar_lock: alarm_count = sum(radar_alarm_status) radar_text = f"雷达警告: {alarm_count}/4" cv2.putText(display_frame, radar_text, (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2) else: # 单画面模式 - 分屏显示 camera_id = display_mode # 创建空白画布 display_frame = np.zeros((SCREEN_HEIGHT, SCREEN_WIDTH, 3), dtype=np.uint8) # 左侧视频区域 with locks[camera_id - 1]: single_frame = frames[camera_id - 1] if single_frame is None or not stream_status[camera_id - 1]: video_frame = create_single_no_signal(camera_id, VIDEO_WIDTH, SCREEN_HEIGHT) else: video_frame = cv2.resize(single_frame, (VIDEO_WIDTH, SCREEN_HEIGHT)) display_frame[0:SCREEN_HEIGHT, 0:VIDEO_WIDTH] = video_frame # 右侧雷达区域 if current_radio_image is not None: radar_display = current_radio_image.copy() # 在雷达图片上叠加距离信息 with radar_lock: distance = radar_distances[camera_id - 1] alarm_status = radar_alarm_status[camera_id - 1] # 显示距离信息 distance_text = f" {distance:.2f}m" if alarm_status: alarm_text = "接近预警!" cv2.putText(radar_display, alarm_text + distance_text, (20, SCREEN_HEIGHT - 20), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 3) display_frame[0:SCREEN_HEIGHT, VIDEO_WIDTH:SCREEN_WIDTH] = radar_display # 显示报警状态 with radar_lock: alarm_count = sum(radar_alarm_status) alarm_text = f"雷达编号: {alarm_count}" cv2.putText(display_frame, alarm_text, (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2) return display_frame def create_no_signal_frame(width, height): frame = np.zeros((height, width, 3), dtype=np.uint8) frame[:] = (40, 40, 40) text = "无信号" font_scale = 1.5 thickness = 3 color = (100, 100, 255) text_size = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, font_scale, thickness)[0] text_x = (width - text_size[0]) // 2 text_y = (height + text_size[1]) // 2 cv2.putText(frame, text, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, font_scale, color, thickness) return frame def create_single_no_signal(camera_id, width, height): frame = np.zeros((height, width, 3), dtype=np.uint8) frame[:] = (40, 40, 40) camera_text = f"相机 {camera_id}" no_signal_text = "无信号" camera_size = cv2.getTextSize(camera_text, cv2.FONT_HERSHEY_SIMPLEX, 1.2, 2)[0] camera_x = (width - camera_size[0]) // 2 camera_y = height // 2 - 40 cv2.putText(frame, camera_text, (camera_x, camera_y), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (100, 255, 100), 2) no_signal_size = cv2.getTextSize(no_signal_text, cv2.FONT_HERSHEY_SIMPLEX, 1.5, 3)[0] no_signal_x = (width - no_signal_size[0]) // 2 no_signal_y = camera_y + 60 cv2.putText(frame, no_signal_text, (no_signal_x, no_signal_y), cv2.FONT_HERSHEY_SIMPLEX, 1.5, (100, 100, 255), 3) status_text = "等待信号接入..." status_size = cv2.getTextSize(status_text, cv2.FONT_HERSHEY_SIMPLEX, 0.7, 2)[0] status_x = (width - status_size[0]) // 2 status_y = no_signal_y + 50 cv2.putText(frame, status_text, (status_x, status_y), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (100, 200, 255), 2) return frame # ---------- 主显示和检测循环 ---------- def main_detection_loop(): global stop_threads, combined_frame, display_mode # 加载雷达图片 if not load_radio_images(): print("Warning: Radar images not loaded properly") # 启动雷达读取器 radar_reader = RadarUARTReader('/dev/ttyS0', 115200) if not radar_reader.start(): print("Warning: Radar reader failed to start") # 加载 anchor try: with open(ANCHORS_FILE, 'r') as f: values = [float(_v) for _v in f.readlines()] anchors = np.array(values).reshape(3, -1, 2).tolist() print(f"Using anchor file: {ANCHORS_FILE}") except: print("Warning: Cannot load anchor file, using default anchors") anchors = [] # 初始化COCO帮助类 co_helper = COCO_test_helper(enable_letter_box=True) # 加载模型 model, platform = setup_model(MODEL_PATH, device_id=0) print("Model loaded, starting detection...") # 设置无边框全屏窗口 if IMG_SHOW: window_name = setup_borderless_fullscreen_window() hide_mouse() # 初始显示画面 initial_frame = np.zeros((SCREEN_HEIGHT, SCREEN_WIDTH, 3), dtype=np.uint8) cv2.putText(initial_frame, "Initializing...", (SCREEN_WIDTH // 2 - 100, SCREEN_HEIGHT // 2), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2) cv2.imshow(window_name, initial_frame) cv2.waitKey(100) frame_counter = 0 detection_cache = { 'boxes': None, 'classes': None, 'scores': None, 'last_frame': -1 } try: print("Entering main display loop...") while not stop_threads: # 根据雷达数据更新显示模式 update_display_mode_by_radar() # 获取最新的合成帧 with combined_lock: current_combined = combined_frame.copy() if combined_frame is not None else None # 只在四分屏模式下进行人体检测 if display_mode == 0 and frame_counter % DETECTION_INTERVAL == 0: if current_combined is not None: boxes, classes, scores = process_frame_for_detection( current_combined, model, platform, co_helper, anchors) detection_cache.update({ 'boxes': boxes, 'classes': classes, 'scores': scores, 'last_frame': frame_counter }) # 创建并显示帧 if IMG_SHOW: try: display_frame = create_display_frame(display_mode, current_combined, detection_cache, frame_counter) cv2.imshow(window_name, display_frame) # 只处理ESC键退出 key = cv2.waitKey(1) & 0xFF if key == 27: # ESC键退出 stop_threads = True break except Exception as e: print(f"Display error: {e}") frame_counter += 1 time.sleep(0.01) finally: if IMG_SHOW: cv2.destroyAllWindows() if hasattr(model, 'release'): model.release() radar_reader.stop() print("Detection loop ended") # ---------- 主函数 ---------- if __name__ == "__main__": os.environ['DISPLAY'] = ':0' print("=" * 60) print("RTSP监控系统 - 简化版本") print("=" * 60) print(f"RTSP URLs:") for i, url in enumerate(RTSP_URLS): print(f" Camera {i + 1}: {url}") print("=" * 60) # 初始化时间戳 current_time = time.time() for i in range(4): stream_last_frame_time[i] = current_time # 启动四个RTSP流读取线程 rtsp_threads = [] for i in range(4): t = threading.Thread(target=read_rtsp_stream, args=(i, RTSP_URLS[i])) t.daemon = True t.start() rtsp_threads.append(t) print(f"✓ 启动RTSP线程 {i}") time.sleep(1) # 错开连接时间 # 启动画面合成线程 combine_thread = threading.Thread(target=combine_frames) combine_thread.daemon = True combine_thread.start() print("✓ 启动画面合成线程") try: # 在主线程中启动检测循环 main_detection_loop() except KeyboardInterrupt: print("用户中断") except Exception as e: print(f"程序错误: {e}") finally: stop_threads = True time.sleep(2) print("所有线程已停止，程序退出")