如何通过子进程实现两路视频流车辆检测结果汇总写入文件？
2026-7-23
问题描述
我有两路视频流，需要汇总两路中检测到的所有车辆总数。目前仅能对单路视频流进行车辆检测，并将检测结果按日期写入cars.txt文件。尝试复制代码实现双路检测时脚本频繁崩溃，该方案不可行。我想尝试使用子进程方案：由第三个脚本启动两个检测脚本、提取检测数值并将汇总结果带日期写入文件。请问该方案是否可行？是否可以直接从脚本获取数值，还是需要脚本输出到控制台再由主脚本读取？
现有检测代码
import cv2 import numpy as np import time import datetime net = cv2.dnn.readNet('weights','cfg') classes = [] with open('data/obj.names','r') as f: classes = f.read().splitlines() cap = cv2.VideoCapture('ipaddress') cap2 = cv2.VideoCapture('ipaddress2') last_time = 0 while True: now = time.time() _,img = cap.read() height, width, _ = img.shape blob = cv2.dnn.blobFromImage(img, 1/255, (416 , 416), (0,0,0) ,swapRB=True,crop=False) net.setInput(blob) output_layer_names = net.getUnconnectedOutLayersNames() layerOutput = net.forward(output_layer_names) boxes = [] car = 0 confidences = [] class_ids =[] for output in layerOutput: for detection in output: scores = detection[5:] class_id = np.argmax(scores) confidence = scores[class_id] if confidence > 0.1: center_x = int(detection[0]*width) center_y = int(detection[1]*height) w = int(detection[2]*width) h = int(detection[3]*height) x = int(center_x - w/2) y = int(center_y - h/2) boxes.append([x,y,w,h]) confidences.append((float(confidence))) class_ids.append(class_id) indexes = cv2.dnn.NMSBoxes(boxes,confidences,0.5,0.4) font = cv2.QT_FONT_NORMAL colors = np.random.uniform(0,255,size=(len(boxes),3)) for i in indexes.flatten(): labels = str(classes[class_ids[i]]) if(labels == 'car'): car+=1 if now - last_time >= 1800: with open('cars.txt','a') as f: f.write(f'{datetime.datetime.now()}: {car}\n') last_time = now for i in indexes.flatten(): x,y,w,h = boxes[i] label =str(classes[class_ids[i]]) confidence = str(round(confidences[i],1)) color = colors[i] cv2.rectangle(img,(x,y),(x+w , y+h), [0,255,0], 2) cv2.putText(img, 'detected cars: ' + str(car), (20, 20), font, 0.8, (0,0,0),2) cv2.imshow('Image',img) key = cv2.waitKey(1) if key == 10: break cap.release() cv2.destroyAllWindows()
解决方案与说明
子进程方案可行性
子进程方案完全可行，这是解决单脚本多流检测崩溃的常用方案。单脚本双路检测崩溃的核心原因是OpenCV视频捕获、DNN推理在单线程下资源占用过高（比如GPU/CPU负载拉满、内存泄漏），拆分到独立子进程后，每个进程单独处理一路流，实现资源隔离，能有效避免崩溃问题。
数值获取方式选择
两种方式都可以实现，具体根据需求选择：
方式1：检测脚本输出到控制台，主脚本读取
这是最简单易实现的方案，无需大幅修改现有代码：
- 修改检测脚本：将原本写入文件的车辆数改为打印到标准输出（比如
print(car)），确保输出内容为纯数字或固定格式（如car:15），避免其他日志干扰。 - 主脚本通过
subprocess.Popen启动两个检测脚本，用stdout=subprocess.PIPE捕获输出，解析出车辆数。 - 主脚本按原定时逻辑（每30分钟）汇总数值并写入文件。
主脚本示例片段
import subprocess import time import datetime # 启动两个检测子进程 proc1 = subprocess.Popen(['python', 'detect_stream1.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) proc2 = subprocess.Popen(['python', 'detect_stream2.py'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) last_summary_time = 0 while True: now = time.time() if now - last_summary_time >= 1800: # 读取子进程输出（需处理缓冲区，避免阻塞） count1 = int(proc1.stdout.readline().strip()) count2 = int(proc2.stdout.readline().strip()) total = count1 + count2 with open('total_cars.txt', 'a') as f: f.write(f'{datetime.datetime.now()}: total cars {total} (stream1: {count1}, stream2: {count2})\n') last_summary_time = now time.sleep(1)
方式2：进程间通信直接获取数值
如果需要更可靠、实时的数值传递，可采用进程间通信（IPC）方案：
- 使用
multiprocessing模块的Queue或Pipe：将检测逻辑封装为函数，主脚本用multiprocessing.Process启动两个检测进程，检测进程将车辆数放入队列，主脚本从队列取数汇总。 - 临时文件：检测脚本定时将车辆数写入独立临时文件（如
stream1_count.txt），主脚本定时读取两个文件的数值汇总。
这种方式比控制台输出更稳定，适合对数据准确性要求高的场景，以下是multiprocessing实现示例：
检测函数（抽离原脚本逻辑）
import cv2 import numpy as np import time from multiprocessing import Queue def detect_stream(ip_address, queue): net = cv2.dnn.readNet('weights','cfg') classes = [] with open('data/obj.names','r') as f: classes = f.read().splitlines() cap = cv2.VideoCapture(ip_address) last_time = 0 while True: now = time.time() _,img = cap.read() if img is None: time.sleep(1) continue height, width, _ = img.shape blob = cv2.dnn.blobFromImage(img, 1/255, (416 , 416), (0,0,0) ,swapRB=True,crop=False) net.setInput(blob) output_layer_names = net.getUnconnectedOutLayersNames() layerOutput = net.forward(output_layer_names) boxes = [] car = 0 confidences = [] class_ids =[] for output in layerOutput: for detection in output: scores = detection[5:] class_id = np.argmax(scores) confidence = scores[class_id] if confidence > 0.1: center_x = int(detection[0]*width) center_y = int(detection[1]*height) w = int(detection[2]*width) h = int(detection[3]*height) x = int(center_x - w/2) y = int(center_y - h/2) boxes.append([x,y,w,h]) confidences.append((float(confidence))) class_ids.append(class_id) indexes = cv2.dnn.NMSBoxes(boxes,confidences,0.5,0.4) if len(indexes) > 0: for i in indexes.flatten(): labels = str(classes[class_ids[i]]) if labels == 'car': car +=1 if now - last_time >= 1800: stream_tag = 'stream1' if 'ipaddress1' in ip_address else 'stream2' queue.put((stream_tag, car)) last_time = now # 保留显示逻辑（可选） font = cv2.QT_FONT_NORMAL cv2.putText(img, 'detected cars: ' + str(car), (20, 20), font, 0.8, (0,0,0),2) cv2.imshow(f'Stream {ip_address}', img) key = cv2.waitKey(1) if key == 10: break cap.release() cv2.destroyAllWindows()
主脚本
import multiprocessing import time import datetime if __name__ == '__main__': queue = multiprocessing.Queue() # 启动两个检测进程 p1 = multiprocessing.Process(target=detect_stream, args=('ipaddress1', queue)) p2 = multiprocessing.Process(target=detect_stream, args=('ipaddress2', queue)) p1.start() p2.start() last_summary_time = 0 counts = {'stream1':0, 'stream2':0} while True: now = time.time() # 读取队列中的检测数据 while not queue.empty(): stream, count = queue.get() counts[stream] = count # 定时汇总写入文件 if now - last_summary_time >= 1800: total = counts['stream1'] + counts['stream2'] with open('total_cars.txt', 'a') as f: f.write(f'{datetime.datetime.now()}: total cars {total} (stream1: {counts["stream1"]}, stream2: {counts["stream2"]})\n') last_summary_time = now time.sleep(1) p1.join() p2.join()
内容的提问来源于stack exchange，提问作者Isshen
相关产品推荐
相关产品推荐
