web端rtmp推拉流测试、抽帧识别计数，一键式生成巡检报告

也想出现在这里？点击联系我~

本文旨在实现无人机城市交通智慧巡检中的一个模块——无人机视频实时推拉流以及识别流并在前端展示，同时，统计目标数量以及违停数量，生成结果评估，一并发送到前端展示。对于本文任何技术上的空缺，可在博主主页前面博客寻找，有任何问题欢迎私信或评论区讨论！！！

涉及技术栈：

Django5+vue3+websocket+SRS+FFmpeg+RTMP+YOLOv8+AI模型+异步+多进程+flv.js+node.js

web端推拉流测试、抽帧识别计数，一键式生成巡检报告

项目结构(Django)：
├── DJI_yolo
│   ├── __init__.py
│   ├── __pycache__
│   ├── asgi.py
│   ├── settings.py
│   ├── templates
│   ├── urls.py
│   └── wsgi.py
├── app01
│   ├── __init__.py
│   ├── __pycache__
│   ├── admin.py
│   ├── apps.py
│   ├── car_best.pt
│   ├── consumers
│   │   ├── __init__.py
│   │   ├── __pycache__
│   │   ├── detection_consumer.py
│   │   └── report_consumer.py
│   ├── migrations
│   │   ├── __init__.py
│   │   └── __pycache__
│   ├── models.py
│   ├── pictures
│   ├── routings.py
│   ├── services
│   │   ├── __init__.py
│   │   ├── __pycache__
│   │   └── detection_engine.py
│   ├── stream_status.py
│   ├── tests.py
│   ├── utils.py
│   └── views.py
├── manage.py
├── media
│   ├── 2025-06-02
│   │   ├── 20时14分54秒-20时17分03秒
│   │   │   ├── 关键帧图片
│   │   │   ├── 原视频
│   │   │   └── 识别视频
│   │   ├── 20时26分11秒-20时29分00秒
│   │   │   ├── 关键帧图片
│   │   │   ├── 原视频
│   │   │   └── 识别视频

存在的问题，亟需解决：

1、后端逐帧识别并返回前端，前端显示卡顿，考虑跳帧识别，一秒识别一帧，剩余原始帧直接返回。

2、曾尝试过使用 yolo 中的 model 类中的 track 方法进行跨帧追踪，为每帧每个目标进行编号（id），提取帧中目标特征（速度、目标面积、唯位移等等），进行特征化工程，从而判断相邻两识别帧中的某两目标是否为同一目标。由于添加追踪算法，单帧识别时间过长，单帧识别时间大于拉流帧传输时间，导致进程堵塞，程序崩溃。后续采取措施改进。

3、模型并未做违停检测，考虑重新训练模型，添加违停训练集，重新训练，顺便搭配 yolo12 环境，用 yolo12 进行训练。

4、Django 后端接受到流，并不能直接开始识别，而是先用四五秒时间加载模型，需要优化。

5、执行任务、完成任务到数据存储、前端显示巡检报告，延迟较高，需要优化。

6、为后端添加车辆运动状态算法，若连续识别的两个帧同一目标静止，并且识别为违停，则判断为违停；若非静止但识别违停，不做处理。相对静止判断？无人机在动，无法建立参考系。

7、进行车牌识别。

代码及粗略解释：

detection_consumer.py：

此处代码使用 FFmpeg 拉取 SRS 上的流，并使用 detection_engine 中的类进行帧识别和计数，然后判断计数，调用讯飞星火 API 生成 AI 答复，获取关键帧，将所有信息封装成一个字典，并传给前端，以用于违停报告生成。同时，将实时识别帧通过 websocket 的消费者传给前端显示。

1. import asyncio
2. import base64
3. import os
4. import subprocess
5. import time
6. import traceback
7. from datetime import datetime
8. from pathlib import Path
9. from .report_consumer import ReportConsumer
10. import numpy as np
11. import cv2
12. from channels.generic.websocket import AsyncWebsocketConsumer
13. from channels.layers import get_channel_layer
14. from app01.services.detection_engine import DetectionEngine, logger
15. from concurrent.futures import ThreadPoolExecutor
16. from app01.utils import rename_time_folder # 导入重命名函数
17. from app01.stream_status import stream_status, status_lock # 从新模块导入
18. import requests
20. RTMP_URL = “rtmp://127.0.0.1:1935/live/stream”
21. channel_layer = get_channel_layer()
23. class DetectionStreamConsumer(AsyncWebsocketConsumer):
24. async def connect(self):
25. await self.channel_layer.group_add(“detection_group”, self.channel_name)
26. await self.accept()
27. print(“WebSocket 连接建立成功”)
29. async def disconnect(self, code):
30. await self.channel_layer.group_discard(“detection_group”, self.channel_name)
31. print(“WebSocket 连接关闭”)
33. async def start_detection(self):
34. print(“开始检测帧流…”)
35. await self.monitor_stream_status()
37. async def monitor_stream_status(self):
38. while True:
39. if stream_status.is_streaming:
40. logger.info(“检测到推流，开始处理…”)
41. await self.start_detection_stream()
42. else:
43. logger.info(“当前无推流，等待中…”)
44. await asyncio.sleep(1)
46. async def start_detection_stream(self, all_dict=None):
47. ffmpeg_pull_command = [
48. “ffmpeg”,
49. “-c:v”, “h264”,
50. “-i”, RTMP_URL,
51. “-f”, “rawvideo”,
52. “-pix_fmt”, “bgr24”,
53. “-s”, “640×360”,
54. “-r”, “5”,
55. “-vcodec”, “rawvideo”,
56. “-an”,
57. “-flags”, “+low_delay”,
58. “-tune”, “zerolatency”,
59. “-“
60. ]
62. try:
63. process = subprocess.Popen(
64. ffmpeg_pull_command,
65. stdout=subprocess.PIPE,
66. stderr=subprocess.DEVNULL,
67. bufsize=10 * 8
68. )
69. except Exception as e:
70. logger.error(f”FFmpeg 子进程启动失败: {e}“)
71. return
73. if process.stdout is None:
74. logger.error(“FFmpeg stdout 为空”)
75. return
77. frame_width, frame_height = 640, 360
78. frame_size = frame_width * frame_height * 3
79. executor = ThreadPoolExecutor(max_workers=4) # 最多可以同时运行 4 个线程
80. engine = DetectionEngine()
81. frame_count = 0
82. skip_interval = 1
84. try:
85. while stream_status.is_streaming:
86. try:
87. loop = asyncio.get_event_loop()
88. raw_frame = await asyncio.wait_for(
89. loop.run_in_executor(executor, process.stdout.read, frame_size),
90. timeout=8 # 超时检测断流
91. )
92. except asyncio.TimeoutError:
93. logger.warning(“读取帧超时，触发流结束”)
94. with status_lock:
95. stream_status.is_streaming = False
96. break
98. if len(raw_frame) != frame_size:
99. continue
101. try:
102. frame = np.frombuffer(raw_frame, dtype=np.uint8).reshape((frame_height, frame_width, 3))
103. except Exception as e:
104. logger.warning(f”帧解析失败: {e}“)
105. continue
107. frame_count += 1
108. if frame_count % skip_interval != 0:
109. continue
111. result = await loop.run_in_executor(executor, engine.process_frame, frame)
112. # 增加空值检查
113. if result is None:
114. logger.warning(“检测结果为空，跳过处理”)
115. continue
116. processed_frame, detected_classes = result
118. # 更新车辆统计（仅按类别）
119. with DetectionEngine.counter_lock:
120. for class_id in detected_classes:
121. if class_id == 0:
122. DetectionEngine.total_count[‘car’] += 1
123. DetectionEngine.total_count[‘total’] += 1
124. elif class_id == 1:
125. DetectionEngine.total_count[‘bus’] += 1
126. DetectionEngine.total_count[‘total’] += 1
127. elif class_id == 2:
128. DetectionEngine.total_count[‘truck’] += 1
129. DetectionEngine.total_count[‘total’] += 1
130. elif class_id == 3:
131. DetectionEngine.total_count[‘van’] += 1
132. DetectionEngine.total_count[‘total’] += 1
134. _, jpeg = cv2.imencode(‘.jpg’, processed_frame, [int(cv2.IMWRITE_JPEG_QUALITY), 50])
135. if channel_layer:
136. await channel_layer.group_send(
137. “detection_group”,
138. {“type”: “send_frame”, “frame”: jpeg.tobytes()}
139. )
140. except Exception as e:
141. logger.error(f”检测处理错误: {e}“)
142. logger.error(traceback.format_exc())
143. finally:
144. logger.warning(“流结束，进入处理…”)
145. if process.returncode is None:
146. process.kill()
147. logger.warning(“FFmpeg 进程已终止”)
149. # 延迟一小段时间，确保文件操作完全释放
150. time.sleep(1)
152. stream_status.is_streaming = False
153. stream_status.end_datetime = datetime.now() # 记录结束时间
155. logger.warning(“正在获取三个关键帧…”)
156. # 找3个关键帧，必须在重命名前执行
157. folder = Path(f”{stream_status.image_path}“)
158. files = [f for f in folder.iterdir() if f.is_file()]
159. if not files:
160. return None, None, None
161. # 按修改时间排序
162. files.sort(key=lambda x: x.stat().st_mtime)
163. first = files[0]
164. last = files[-1]
165. middle = files[len(files) // 2]
167. # 转换为Base64
168. stream_status.image_li = [
169. self.image_to_base64(str(f)) # 传入字符串路径，避免Path对象序列化问题
170. for f in [first, middle, last]
171. ]
173. logger.warning(“正在重命名文件夹…”)
174. # 先复制 time_path，因为后续可能被修改
175. time_path = stream_status.time_path
176. start_datetime = stream_status.start_datetime
177. end_datetime = stream_status.end_datetime
178. save_path = “”
179. # 调用重命名函数（使用完整 datetime）
180. if time_path and start_datetime:
182. path = rename_time_folder(
183. time_path,
184. start_datetime, # 传入开始时间
185. end_datetime # 传入结束时间
186. )
187. save_path = path
188. logger.warning(
189. f”文件夹已重命名为 {start_datetime.strftime(‘%H时%M分%S秒’)}–{end_datetime.strftime(‘%H时%M分%S秒’)}“)
191. try:
192. logger.warning(“正在更新数据…”)
193. # 更新识别结果
194. stats = self.correct_overcounted_stats(DetectionEngine.total_count)
195. print(“stats:”, stats)
196. stream_status.total_stats = stats
197. print(“stream_status.total_stats：”, stream_status.total_stats)
198. self.save_vehicle_stats(stats, save_path)
200. except Exception as e:
201. logger.error(f”保存统计信息失败: {e}“)
202. print(“正在获取AI答复…”)
203. self.AI_response()
205. # 定义总数据，并序列化
206. all_dict = {
207. ‘datetime’: stream_status.start_time.strftime(“%Y-%m-%d %H:%M:%S.%f”), # 序列化，转为字符串
208. ‘image_li’: [bs for bs in stream_status.image_li], # 关键帧转二进制
209. ‘detect_car’: [_ for _ in stream_status.total_stats.values()],
210. ‘Al_response’: stream_status.AI_talk,
211. }
213. print(“all_dict:”, all_dict)
214. print(“正在发送信息…”)
215. await ReportConsumer.send_report_data(all_dict)
217. executor.shutdown(wait=False)
218. DetectionEngine.reset_stats()
219. stream_status.reset_all_dict()
221. def save_vehicle_stats(self, stats, save_path=None):
222. # 如果未提供路径，使用默认的 time_path
223. if not save_path:
224. save_path = stream_status.time_path
226. if not save_path:
227. logger.warning(“无法保存统计信息：路径为空”)
228. return
230. stats_path = os.path.join(save_path, ‘vehicle_stats.txt’)
231. try:
232. with open(stats_path, ‘w’, encoding=‘utf-8-sig’) as f:
233. f.write(“车辆统计结果\n”)
234. f.write(“————————-\n”)
235. for k, v in stats.items():
236. f.write(f”{k}: {v}\n”)
237. f.write(“————————-\n”)
238. logger.info(f”统计信息已保存至 {stats_path}“)
239. except Exception as e:
240. logger.error(f”写入统计信息失败: {e}“)
243. def AI_response(self):
244. “””获取AI回复”””
245. url = “https://spark-api-open.xf-yun.com/v1/chat/completions”
246. headers = {
247. “Authorization”: “Bearer pDLDvpUbFDTEQZACQSjD:CqZewebAdgpuvxrVbbAv”,
248. “Content-Type”: “application/json”,
249. }
250. content = (
251. f’我现在要做一个无人机城市交通智慧巡检报告，现在巡检结果是：巡检时间：{stream_status.start_time}‘
252. f’小汽车有{stream_status.total_stats[“car”]}辆，面包车有{stream_status.total_stats[“van”]}辆,’
253. f’公交车有{stream_status.total_stats[“bus”]}辆，卡车有{stream_status.total_stats[“truck”]}辆,’
254. f’识别到的违停车辆共有{stream_status.total_stats[“illegally_car”]}辆。’
255. f’帮我生成一段结论，要求不要有废话，也不要写具体那四个类别识别到的车辆数，字数200字，语言精炼严谨’
256. )
258. question = {
259. “role”: “user”,
260. “content”: content
261. }
263. data = {
264. “max_tokens”: 4096,
265. “top_k”: 4,
266. “messages”: [question],
267. “temperature”: 0.5,
268. “model”: “4.0Ultra”,
269. “stream”: False,
270. }
271. response = requests.post(url, headers=headers, json=data)
272. response.raise_for_status()
273. result = response.json()
275. # 提取模型输出的内容
276. reply = result.get(‘choices’, [{}])[0].get(‘message’, {}).get(‘content’, ”).strip()
277. print(reply)
278. stream_status.AI_talk = reply
280. # 读取图片文件并转换为 Base64
281. @staticmethod # 静态方法不应接收self参数
282. def image_to_base64(image_path):
283. with open(image_path, “rb”) as img_file:
284. base64_str = base64.b64encode(img_file.read()).decode(“utf-8”)
285. return f”data:image/jpeg;base64,{base64_str}“
287. @staticmethod # 去掉self第一个参数
288. def correct_overcounted_stats(stats, avg_appearance=6):
289. “””
290. 简单修正严重重复统计的结果
291. “””
292. corrected = {}
293. for cls in [‘car’, ‘van’, ‘bus’, ‘truck’]:
294. if cls in stats:
295. corrected[cls] = max(0, int(round(stats[cls] / avg_appearance)))
297. # 可选：重新计算 total
298. corrected[‘total’] = sum(corrected.values())
299. corrected[‘illegally_car’] = int(corrected[‘total’] / 100)
300. print(corrected)
302. return corrected
304. async def send_frame(self, event):
305. frame_bytes = event[“frame”]
306. await self.send(bytes_data=frame_bytes)

     stream_status.py：

     本文件用于 Django 项目全局状态管理，相当于 vue3 中的 pinia ， is_streaming 是前端推本地视频、后端上传 SRS 后，变为 True ，后端因为 is_streaming = True 而开始拉流，当超过 6 秒未拉到流，判断流结束，is_streaming = False。还定义了全局状态锁、状态重置方法、时间获取方法、文件夹名获取方法、状态实例。全局状态锁是为了防止多线程修改同一个共享变量，只有with status_lock：在 with 块内，只有当前线程持有锁时才能运行代码，其他线程必须等待，相当于在此处把锁”锁“起来了。

     1. import threading
     3. # 全局状态锁和状态对象
     4. status_lock = threading.Lock()
     5. class StreamStatus:
     7. def __init__(self):
     8. self.is_streaming = False # 是否正在推流/检测
     9. self.start_time = None # 推流开始时间（datetime对象）
     10. self.end_time = None # 推流结束时间（datetime对象）
     11. self.time_path = None # 当前时间文件夹路径（如 Media/2025-06-01/12点）
     12. self.total_stats = { # 车辆统计
     13. “car”: 0,
     14. “van”: 0,
     15. “bus”: 0,
     16. “truck”: 0,
     17. “total”: 0,
     18. “illegally_car”: 0,
     19. }
     20. self.last_frame_time = None # 最后收到有效帧的时间（用于断流检测）
     21. self.image_path = None
     22. self.image_li = []
     23. self.AI_talk = None
     24. self.all_dict = {}
     26. @property
     27. def date_folder(self):
     28. “””获取日期文件夹名称（如 2025-06-01）”””
     29. if self.start_time:
     30. return self.start_time.strftime(“%Y-%m-%d”)
     31. return None
     33. @property
     34. def start_hour(self):
     35. “””获取开始小时（如 12）”””
     36. if self.start_time:
     37. return self.start_time.hour
     38. return None
     40. def reset(self):
     41. “””重置状态（用于新任务开始）”””
     42. self.is_streaming = False
     43. self.start_time = None
     44. self.end_time = None
     45. self.time_path = None
     46. self.total_stats = {k: 0 for k in self.total_stats}
     47. self.last_frame_time = None
     49. def reset_all_dict(self):
     50. self.last_frame_time = None # 最后收到有效帧的时间（用于断流检测）
     51. self.image_path = None
     52. self.image_li = []
     53. self.AI_talk = None
     54. self.all_dict = {}
     55. self.all_dict = {}
     56. # 全局唯一的状态实例
     57. stream_status = StreamStatus()

      

     AI写代码python

     detection_engine.py：

     此处是 detection_consumer.py 使用的服务代码块，用于为其提供帧识别服务，consumer 拉到的每一个流都会调用该类中的 process_frame 类方法，并同时进行计数和关键帧保存

     1. import threading
     2. import time
     3. import cv2
     4. from ultralytics import YOLO
     5. import logging
     6. import os
     7. from app01.stream_status import stream_status, status_lock # 从新模块导入
     9. logger = logging.getLogger(__name__)
     11. MODEL_PATH = “F:\\FullStack\\Django\\DJI_yolo\\app01\\car_best.pt”
     12. model = YOLO(MODEL_PATH)
     14. # 关键帧存储位置
     15. pictures_dir = “pictures”
     16. if not os.path.exists(pictures_dir):
     17. os.makedirs(pictures_dir)
     20. class DetectionEngine:
     21. frame_counter = 1
     22. counter_lock = threading.Lock() # 线程锁，保证计数器安全
     24. # 新增：全局计数器和已统计 ID
     25. total_count = {
     26. ‘car’: 0,
     27. ‘van’: 0,
     28. ‘bus’: 0,
     29. ‘truck’: 0,
     30. “total”: 0,
     31. “illegally_car”: 0
     32. }
     33. # 下次任务重置
     34. @classmethod
     35. def reset_stats(cls):
     36. with cls.counter_lock:
     37. for k in cls.total_count:
     38. cls.total_count[k] = 0
     40. # 下次任务重置
     41. @classmethod
     42. def update_stats(cls, vehicle_type):
     43. with cls.counter_lock:
     44. if vehicle_type in cls.total_count:
     45. cls.total_count[vehicle_type] += 1
     46. cls.total_count[“total”] += 1
     48. def __init__(self):
     49. self.model = model
     50. # 调整检测参数，减少计算量
     51. self.detect_params = {
     52. ‘conf’: 0.3, # 提高置信度阈值，减少无效检测
     53. ‘iou’: 0.5, # 调整 NMS 阈值
     54. ‘imgsz’: [384, 640], # 输入尺寸，降低分辨率
     55. ‘classes’: [0, 1, 2, 3], # 仅检测车辆类别（2: car, 5: bus, 7: truck）
     56. ‘device’: 0, # 使用 GPU
     57. ‘half’: True, # 使用 FP16 精度
     58. ‘show’: False # 关闭实时显示，减少开销
     59. }
     61. def process_frame(self, frame):
     62. “””处理单帧图像：YOLO推理 + 保存结果到关键帧图片文件夹”””
     63. results = model(frame)
     64. detected_objects = []
     65. annotated_frame = frame.copy() # 复制原始帧以防止修改原图
     66. # 防御性检查：确保 results 非空且包含有效数据
     67. if not results or len(results) == 0:
     68. logger.warning(“未获取到检测结果，跳过处理”)
     69. return frame, []
     71. boxes = results[0].boxes
     72. if boxes is None or len(boxes) == 0:
     73. logger.warning(“检测结果为空，跳过处理”)
     74. return frame, []
     75. try:
     76. # 假设类别信息存储在 cls 属性中
     77. classes = boxes.cls.int().cpu().tolist() if hasattr(boxes, ‘cls’) and boxes.cls is not None else []
     79. for class_id in classes:
     80. detected_objects.append(class_id)
     82. annotated_frame = results[0].plot() # 绘制检测框
     83. except Exception as e:
     84. logger.error(f”解析检测数据失败: {e}“)
     85. return annotated_frame, []
     86. with status_lock:
     87. car_count = sum(1 for cls in detected_objects if cls == 0)
     88. if car_count > 18:
     89. try:
     90. timestamp = int(time.time())
     91. save_path = os.path.join(stream_status.image_path, f”keyframe_{timestamp}.jpg”)
     92. cv2.imwrite(save_path, annotated_frame)
     93. logger.info(f”关键帧保存成功: {save_path}“)
     94. except Exception as e:
     95. logger.error(f”保存关键帧失败: {e}“)
     97. return annotated_frame, detected_objects

      

     repoet_consumer.py：

     用于将前面 detection_consumer.py 中的字典数据发送给前端，首先会将数据缓存到后端，等到前端消费者建立成功，立马发送出去。

     1. import asyncio
     3. from channels.generic.websocket import AsyncWebsocketConsumer
     4. from channels.layers import get_channel_layer
     5. import json
     7. # 缓存队列：用于保存尚未发送的数据
     8. queue = []
     9. # 记录是否有消费者连接
     10. has_consumer = False
     12. class ReportConsumer(AsyncWebsocketConsumer):
     13. async def connect(self):
     14. global has_consumer
     15. # 加入指定组
     16. await self.channel_layer.group_add(“report_group”, self.channel_name)
     17. await self.accept()
     18. print(“巡检报告WebSocket 已连接”)
     19. # 设置有消费者连接的标志
     20. has_consumer = True
     21. # 尝试发送缓存数据
     22. await flush_queue(self.channel_layer)
     24. async def disconnect(self, close_code):
     25. global has_consumer
     26. # 移除组成员
     27. await self.channel_layer.group_discard(“report_group”, self.channel_name)
     28. print(“巡检报告WebSocket 断开连接”)
     29. # 重置消费者连接标志
     30. has_consumer = False
     32. async def receive(self, text_data=None, bytes_data=None):
     33. pass
     35. async def send_report(self, event):
     36. “””处理 send_report 类型的消息并发送给客户端”””
     37. data = event[‘data’]
     38. print(“【发送到客户端】”, data)
     40. # 转换为JSON字符串发送
     41. try:
     42. # 处理可能包含的非JSON可序列化对象
     43. # 这里假设 image_li 包含 Path 对象，需要转换为字符串
     44. if ‘image_li’ in data:
     45. data[‘image_li’] = [str(path) for path in data[‘image_li’]]
     47. await self.send(text_data=json.dumps(data))
     48. except Exception as e:
     49. print(f”【发送到客户端失败】{e}“)
     51. @classmethod
     52. async def send_report_data(cls, data):
     53. “””供其他模块调用的方法”””
     54. global queue, has_consumer
     55. # 不再检查是否有人连接，直接发送或缓存,生产模式用Redis
     56. channel_layer = get_channel_layer()
     58. # 如果没有消费者连接，将数据加入队列
     59. if not has_consumer:
     60. queue.append(data)
     61. print(f”【数据已缓存】等待消费者连接: {data}“)
     62. else:
     63. # 有消费者连接，直接发送
     64. await send_report(channel_layer, data)
     67. # ========== 全局函数 ==========
     68. # 定期检查并发送缓存数据的任务
     69. async def queue_check_task():
     70. channel_layer = get_channel_layer()
     71. while True:
     72. await asyncio.sleep(1) # 每秒检查一次
     73. if queue and has_consumer:
     74. await flush_queue(channel_layer)
     75. # 在应用启动时启动队列检查任务
     76. async def start_queue_check():
     77. asyncio.create_task(queue_check_task())
     78. async def send_report(channel_layer, data):
     79. try:
     80. await channel_layer.group_send(
     81. “report_group”,
     82. {
     83. “type”: “send_report”,
     84. “data”: data,
     85. },
     86. )
     87. print(“【发送成功】”, data)
     88. except Exception as e:
     89. print(f”【发送失败】{e}“)
     92. async def flush_queue(channel_layer):
     93. global queue
     94. if not queue:
     95. return
     97. items = list(queue)
     98. queue.clear()
     99. for data in items:
     100. await send_report(channel_layer, data)

      

     routings.py：

     用于定义 websocket 的后端接收路径，并连接消费者

     1. from django.urls import re_path
     2. from app01.consumers import detection_consumer, report_consumer
     4. websocket_urlpatterns = [
     5. re_path(r’^ws/detection/$’, detection_consumer.DetectionStreamConsumer.as_asgi()),
     6. re_path(r’^ws/report/$’, report_consumer.ReportConsumer.as_asgi()),
     7. ]

      

     urls.py：

     用于接收前端发送的 http 请求，并与 view.py 中的视图函数建立联系

     1. “””
     2. URL configuration for DJI_yolo project.
     4. The `urlpatterns` list routes URLs to views. For more information please see:
     5. https://docs.djangoproject.com/en/5.1/topics/http/urls/
     6. Examples:
     7. Function views
     8. 1. Add an import: from my_app import views
     9. 2. Add a URL to urlpatterns: path(”, views.home, name=’home’)
     10. Class-based views
     11. 1. Add an import: from other_app.views import Home
     12. 2. Add a URL to urlpatterns: path(”, Home.as_view(), name=’home’)
     13. Including another URLconf
     14. 1. Import the include() function: from django.urls import include, path
     15. 2. Add a URL to urlpatqterns: path(‘blog/’, include(‘blog.urls’))
     16. “””
     17. from django.conf.urls.static import static
     18. from django.contrib import admin
     19. from django.urls import path
     21. from DJI_yolo import settings
     22. from app01 import views
     23. urlpatterns = [
     24. path(“admin/”, admin.site.urls),
     25. path(‘csrf/’, views.get_csrf_token, name=‘get_csrf_token’),
     26. # 推流测试
     27. path(‘push_stream/’, views.push_stream, name=‘push_stream’),
     29. # 媒体库管理
     30. path(‘api4/date-folders/’, views.get_date_folders, name=‘date-folders’),
     31. path(‘api4/time-folders/<str:date>/’, views.get_time_folders, name=‘time-folders’),
     32. path(‘api4/files/<str:date>/<str:time_range>/<str:category>/count/’, views.get_file_count, name=‘file-count’),
     33. path(‘api4/files/<str:date>/<str:time_range>/<str:category>/’, views.get_files, name=‘files’),
     34. path(‘api4/delete-file/<str:date>/<str:time_range>/<str:category>/<str:filename>/’, views.delete_file,
     35. name=‘delete-file’),
     36. path(‘delete-file/<str:date>/<str:time_range>/<str:category>/<str:filename>’, views.delete_file,
     37. name=‘delete-file’), # 添加一个用于删除文件的 URL 路径
     39. ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

      

     views.py：

     用于处理前端请求，这里报告推流、媒体管理系统、原始视频存储逻辑等等。

     1. # views.py
     2. from datetime import datetime
     3. from pathlib import Path
     4. import urllib.parse
     5. from django.middleware.csrf import get_token
     6. import os
     7. import subprocess
     8. import threading
     9. from django.http import JsonResponse
     10. from django.conf import settings
     11. import asyncio
     12. from .consumers.detection_consumer import DetectionStreamConsumer
     13. from .utils import create_date_folder, create_time_folder
     14. from .stream_status import stream_status
     16. MEDIA_ROOT = Path(settings.MEDIA_ROOT)
     18. def get_csrf_token(request):
     19. token = get_token(request)
     20. print(token)
     21. return JsonResponse({‘csrfToken’: token})
     23. def push_stream(request):
     24. print(“前端返回本地视频，准备推流”)
     25. video_file = request.FILES.get(‘video’)
     26. if not video_file:
     27. return JsonResponse({‘error’: ‘未上传视频文件’}, status=400)
     29. # ——————- 创建精确时间文件夹 ——————-
     30. start_datetime = datetime.now() # 记录精确到秒的开始时间
     32. date_str = start_datetime.strftime(‘%Y-%m-%d’)
     33. date_path = create_date_folder(date_str)
     34. stream_status.start_time = datetime.now()
     35. # 创建带时分秒的文件夹（如 15:30:45）
     36. time_path, time_folder = create_time_folder(date_path, start_datetime)
     38. # 创建三类子文件夹
     39. original_path = os.path.join(time_path, ‘原视频’)
     40. recognized_path = os.path.join(time_path, ‘识别视频’)
     41. image_path = os.path.join(time_path, ‘关键帧图片’)
     42. stream_status.image_path = image_path # 保存到全局状态
     43. for folder in [original_path, recognized_path, image_path]:
     44. os.makedirs(folder, exist_ok=True)
     46. # ——————- 保存上传视频 ——————-
     47. original_video_name = f”original_{start_datetime.strftime(‘%H%M%S’)}.mp4″
     48. original_video_path = os.path.join(original_path, original_video_name)
     49. with open(original_video_path, ‘wb’) as f:
     50. for chunk in video_file.chunks():
     51. f.write(chunk)
     53. # ——————- 推流配置 ——————-
     54. push_url = “rtmp://127.0.0.1:1935/live/stream”
     55. recognized_video_name = f”recognized_{start_datetime.strftime(‘%H%M%S’)}.mp4″
     56. recognized_video_path = os.path.join(recognized_path, recognized_video_name)
     59. command = [
     60. “ffmpeg”,
     61. “-re”,
     62. “-i”, original_video_path,
     63. “-c:v”, “libx264”,
     64. “-preset”, “ultrafast”,
     65. “-tune”, “zerolatency”,
     66. “-g”, “50”,
     67. “-r”, “30”,
     68. “-an”,
     69. “-f”, “flv”,
     70. push_url,
     71. “-f”, “mp4”,
     72. recognized_video_path
     73. ]
     75. try:
     76. process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     78. # 保存完整时间状态
     79. stream_status.is_streaming = True
     80. stream_status.start_datetime = start_datetime # 保存完整 datetime
     81. stream_status.time_path = time_path
     82. stream_status.recognized_path = recognized_path
     83. stream_status.image_path = image_path
     85. detection_thread = threading.Thread(
     86. target=run_detection_in_background,
     87. daemon=True
     88. )
     89. detection_thread.start()
     91. return JsonResponse({
     92. ‘status’: ‘success’,
     93. ‘date_folder’: date_str,
     94. ‘time_folder’: time_folder # 格式如 15:30:45
     95. })
     96. except Exception as e:
     97. return JsonResponse({‘error’: str(e)}, status=500)
     99. def run_detection_in_background():
     100. try:
     101. loop = asyncio.new_event_loop()
     102. asyncio.set_event_loop(loop)
     103. loop.run_until_complete(DetectionStreamConsumer().start_detection())
     104. except Exception as e:
     105. print(f”后台检测线程异常: {e}“)
     107. def get_date_folders(request):
     108. page = int(request.GET.get(‘page’, 1))
     109. page_size = int(request.GET.get(‘page_size’, 10))
     110. MEDIA_ROOT_PATH = Path(r’F:\FullStack\Django\DJI_yolo\media’)
     112. date_folders = []
     113. for name in os.listdir(MEDIA_ROOT_PATH):
     114. folder_path = MEDIA_ROOT_PATH / name
     115. if folder_path.is_dir() and len(name.split(‘-‘)) == 3:
     116. date_folders.append({‘date’: name})
     118. date_folders.sort(key=lambda x: x[‘date’], reverse=True)
     119. start = (page – 1) * page_size
     120. end = start + page_size
     121. return JsonResponse({
     122. ‘data’: date_folders[start:end],
     123. ‘total’: len(date_folders)
     124. })
     126. def get_time_folders(request, date):
     127. date_path = Path(r’F:\FullStack\Django\DJI_yolo\media’) / date
     128. if not date_path.is_dir():
     129. return JsonResponse({‘error’: ‘日期文件夹不存在’}, status=404)
     131. page = int(request.GET.get(‘page’, 1))
     132. page_size = int(request.GET.get(‘page_size’, 10))
     133. time_folders = []
     135. for name in os.listdir(date_path):
     136. time_path = date_path / name
     137. if time_path.is_dir():
     138. original_count = len(os.listdir(time_path / ‘原视频’)) if (time_path / ‘原视频’).is_dir() else 0
     139. recognized_count = len(os.listdir(time_path / ‘识别视频’)) if (time_path / ‘识别视频’).is_dir() else 0
     140. image_count = len(os.listdir(time_path / ‘关键帧图片’)) if (time_path / ‘关键帧图片’).is_dir() else 0
     142. time_folders.append({
     143. ‘time_range’: name, # 直接使用时间区间名称
     144. ‘original_count’: original_count,
     145. ‘recognized_count’: recognized_count,
     146. ‘image_count’: image_count
     147. })
     149. time_folders.sort(key=lambda x: x[‘time_range’], reverse=True)
     150. start = (page – 1) * page_size
     151. end = start + page_size
     152. return JsonResponse({
     153. ‘data’: time_folders[start:end],
     154. ‘total’: len(time_folders)
     155. })
     157. def get_file_count(request, date, time_range, category):
     158. category_path = Path(r’F:\FullStack\Django\DJI_yolo\media’) / date / time_range / category
     159. if not category_path.is_dir():
     160. return JsonResponse({‘count’: 0})
     162. count = len(os.listdir(category_path))
     163. return JsonResponse({‘count’: count})
     165. def get_files(request, date, time_range, category):
     166. try:
     167. decoded_time_range = urllib.parse.unquote(time_range)
     168. category_path = Path(r’F:\FullStack\Django\DJI_yolo\media’) / date / decoded_time_range / category
     170. if not category_path.is_dir():
     171. return JsonResponse({‘error’: ‘文件类别不存在’}, status=404)
     173. page = int(request.GET.get(‘page’, 1))
     174. page_size = int(request.GET.get(‘page_size’, 10))
     175. files = []
     177. for filename in os.listdir(category_path):
     178. file_path = category_path / filename
     179. if file_path.is_file():
     180. encoded_time_range = urllib.parse.quote(decoded_time_range)
     181. files.append({
     182. ‘name’: filename,
     183. ‘url’: f’http://{request.get_host()}/media/{date}/{encoded_time_range}/{category}/{filename}‘
     184. })
     186. start = (page – 1) * page_size
     187. end = start + page_size
     188. return JsonResponse({
     189. ‘data’: files[start:end],
     190. ‘total’: len(files),
     191. ‘status’: ‘success’
     192. })
     193. except Exception as e:
     194. print(f”文件列表错误: {str(e)}“)
     195. return JsonResponse({
     196. ‘error’: ‘服务器内部错误’,
     197. ‘detail’: str(e)
     198. }, status=500)
     200. def delete_file(request, date, time_range, category, filename):
     201. try:
     202. decoded_time_range = urllib.parse.unquote(time_range)
     203. file_path = Path(r’F:\FullStack\Django\DJI_yolo\media’) / date / decoded_time_range / category / filename
     204. if not file_path.exists():
     205. return JsonResponse({‘error’: ‘文件不存在’}, status=404)
     207. os.remove(file_path)
     208. return JsonResponse({‘status’: ‘success’})
     209. except Exception as e:
     210. return JsonResponse({‘error’: str(e)}, status=500)

      

     AI写代码python

     运行