From df8ad952cc48bee4f0cf15f4f9ff53750b1a745c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E5=96=B5?= Date: Mon, 15 Jun 2026 11:21:35 +0800 Subject: [PATCH] =?UTF-8?q?JPEG=E8=BD=AC=E6=8D=A2=E4=B8=BAH.264(=E9=9B=86?= =?UTF-8?q?=E6=88=90=E5=9C=A8=E6=8E=A5=E6=94=B6=E8=84=9A=E6=9C=AC=E4=B8=AD?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- base_test/receiver.py | 202 +++++++++++++++++++++++++++++++++++------- 1 file changed, 172 insertions(+), 30 deletions(-) diff --git a/base_test/receiver.py b/base_test/receiver.py index 6c36973..8f4c0a2 100644 --- a/base_test/receiver.py +++ b/base_test/receiver.py @@ -5,29 +5,168 @@ import threading import queue import time import os +import sys +import logging -# ==================== 地面站核心配置 ==================== +# ===================================================================== +# 环境初始化：必须在 import av 之前将 imageio-ffmpeg 路径注入 PATH +# ===================================================================== +try: + import imageio_ffmpeg + + # 获取已经存在的 ffmpeg.exe 路径 + ffmpeg_exe_path = imageio_ffmpeg.get_ffmpeg_exe() + ffmpeg_bin_dir = os.path.dirname(ffmpeg_exe_path) + + # 将该文本动态挂载到系统临时环境变量，确保 PyAV 能够识别并调用 + if ffmpeg_bin_dir not in os.environ["PATH"]: + os.environ["PATH"] = ffmpeg_bin_dir + os.pathsep + os.environ["PATH"] + print(f"✅ 成功挂载本地环境 FFmpeg: {ffmpeg_bin_dir}") +except ImportError: + print("❌ 错误：未检测到 imageio_ffmpeg 库，请先执行 `pip install imageio-ffmpeg`") + sys.exit(1) + +import av + +# 配置标准日志 +logging.basicConfig(level=logging.INFO, format='%(asctime)s - [%(levelname)s] - %(message)s') + + +class JPEG2H264Streamer: + """ + 📡 地面端无人机图传转码代理类 (纯内存推流优化版) + + 优化点：不再自己建立 UDP 接收端，而是暴露 push_frame 接口， + 直接接收主线程已经解码完成的 OpenCV BGR 矩阵，消除重复解码，极大降低 CPU 负担。 + """ + + def __init__(self, qgc_ip="127.0.0.1", qgc_port=5600, width=640, height=480, fps=24): + self.logger = logging.getLogger("ProxyStreamer") + self.qgc_ip = qgc_ip + self.qgc_port = qgc_port + self.width = width + self.height = height + self.fps = fps + + # 帧缓冲区（只保留最新图像，防止网络抖动导致的画面累积延迟） + self.frame_queue = queue.Queue(maxsize=3) + self.is_running = False + self.encoder_thread = None + + def start(self): + """ 启动异步 H.264 编码推流线程 """ + self.is_running = True + self.encoder_thread = threading.Thread(target=self._encoder_push_worker, daemon=True) + self.encoder_thread.start() + self.logger.info(f"🚀 H.264 转码推流引擎启动：目标 QGC -> {self.qgc_ip}:{self.qgc_port}") + + def push_frame(self, frame): + """ 供外部（主线程）调用的纯内存图像注入接口 """ + if not self.is_running: + return + try: + self.frame_queue.put_nowait(frame) + except queue.Full: + try: + self.frame_queue.get_nowait() # 弹出老帧 + self.frame_queue.put_nowait(frame) + except queue.Empty: + pass + + def _encoder_push_worker(self): + """ 高性能 H.264 编码与推流核心进程 """ + output_url = f"udp://{self.qgc_ip}:{self.qgc_port}" + output_container = None + stream = None + + pts = 0 + frame_count = 0 + + while self.is_running: + try: + # 初始化/重连机制 + if output_container is None: + output_container = av.open(output_url, mode='w', format='mpegts') + stream = output_container.add_stream('libx264', rate=self.fps) + stream.width = self.width + stream.height = self.height + stream.pix_fmt = 'yuv420p' + + # 💡 极其关键的低延迟高级调优参数 + stream.options = { + 'preset': 'ultrafast', + 'tune': 'zerolatency', + 'g': str(self.fps) # 每一秒强制发一个 I 帧 + } + self.logger.info("PyAV H.264 极速低延迟编码器就绪") + + # 从队列获取主线程传过来的 OpenCV BGR 帧 + try: + frame = self.frame_queue.get(timeout=1.0) + except queue.Empty: + continue + + # 确保分辨率对齐编码器 + if frame.shape[1] != self.width or frame.shape[0] != self.height: + frame = cv2.resize(frame, (self.width, self.height)) + + # 转换色彩空间 (BGR -> RGB) 并封装为 PyAV 帧 + frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) + av_frame = av.VideoFrame.from_numpy_array(frame_rgb, format='rgb24') + + # 分配高精度时钟戳 + av_frame.pts = pts + pts += int(90000 / self.fps) + + # 执行 H.264 实体压缩编码，并注入 UDP 发送端口 + for packet in stream.encode(av_frame): + output_container.mux(packet) + + frame_count += 1 + if frame_count % 100 == 0: + self.logger.info(f"📊 QGC 图传统计: 已成功向 QGC 转发 {frame_count} 帧 H.264 视频") + + except Exception as e: + self.logger.error(f"转码或推流异常（准备重连）: {e}") + output_container = None + time.sleep(1.0) + + # 资源释放 + if output_container: + try: + for packet in stream.encode(): + output_container.mux(packet) + output_container.close() + except Exception: + pass + + def stop(self): + """ 安全关闭代理服务 """ + self.is_running = False + if self.encoder_thread: + self.encoder_thread.join(timeout=2.0) + self.logger.info("转码代理已安全关闭。") + + +# ==================== 地面站全局配置 ==================== PORT = 9999 SAVE_DIR = "../asset/saved_video" SAVE_PATH = os.path.join(SAVE_DIR, "ground_station_record.mp4") -RECORD_FPS = 24.0 # 录像固定的标准帧率 -# ======================================================= +RECORD_FPS = 24.0 # 录像与转码统一固定的标准帧率 -# 确保保存视频的文件夹存在 if not os.path.exists(SAVE_DIR): os.makedirs(SAVE_DIR) -# 核心线程共享：最新的一帧画面与互斥锁（防止读写冲突） +# 核心线程共享区 latest_frame = None frame_lock = threading.Lock() is_running = True def udp_receiver_thread(): - """ 📡 后台极速收包线程 """ + """ 📡 后台唯一极速收包线程 """ global latest_frame, is_running sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - # 放大系统内核缓冲区到 4MB，全力应对图传电台的无线网络抖动 sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4 * 1024 * 1024) try: @@ -45,7 +184,6 @@ def udp_receiver_thread(): frame = cv2.imdecode(data, cv2.IMREAD_COLOR) if frame is not None: - # 安全地更新最新的一帧，瞬间覆盖老帧，杜绝延迟积压 with frame_lock: latest_frame = frame except Exception as e: @@ -59,25 +197,34 @@ def udp_receiver_thread(): def main(): global latest_frame, is_running - # 1. 启动后台无线电收包线程 + # 1. 启动唯一的后台接收线程 recv_t = threading.Thread(target=udp_receiver_thread, daemon=True) recv_t.start() - # 2. 初始化地面站专用录像机 (对应香橙派最后一刻 resize 后的 640x480) + # 2. 初始化本地录像机 fourcc = cv2.VideoWriter_fourcc(*'mp4v') video_writer = cv2.VideoWriter(SAVE_PATH, fourcc, RECORD_FPS, (640, 480)) - print(f"📹 视频录像机就绪，文件将安全保存在: {SAVE_PATH}") + print(f"📹 视频录像机就绪，文件保存在: {SAVE_PATH}") - # 用于计算接收端实时刷新率 + # 3. 启动地面站高性能 H.264 代理组件 (对齐 24 FPS) + proxy = JPEG2H264Streamer( + qgc_ip="127.0.0.1", + qgc_port=5600, + width=640, + height=480, + fps=int(RECORD_FPS) + ) + proxy.start() + + # FPS 统计变量 fps_start_time = time.time() fps_counter = 0 fps_display = 0 - # 严格按照发送端的帧率控制循环步长（1 / 24秒 ≈ 0.0416秒） frame_interval = 1.0 / RECORD_FPS last_frame_recorded = None - print("🖥️ 监视窗口已建立，等待无人机连接推流...") + print("🖥️ 监视窗口已建立，等待无人机连接...") try: while is_running: @@ -88,54 +235,49 @@ def main(): with frame_lock: if latest_frame is not None: current_frame = latest_frame.copy() + latest_frame = None # 取出后置空，防止下一轮循环读到未更新的旧重复帧 - # 无线图传时延容错核心逻辑： + # 无线图传丢包容错核心 if current_frame is not None: - # 拿到了图传新帧，更新最后记录的有效帧 last_frame_recorded = current_frame elif last_frame_recorded is not None: - # 【抗干扰核心】当前瞬间电台丢包没收到新图，复制上一帧顶替，防止视频时间轴缩水变快 current_frame = last_frame_recorded.copy() - # 如果已经有画面可渲染（无论是新帧还是补偿帧） if current_frame is not None: - # 录像写入（PC 的强悍 CPU 在后台无感硬写磁盘，完全不拖累香橙派的 NPU） + # ─── 核心修改点：同时向 QGC 编码流中注入当前帧 ─── + proxy.push_frame(current_frame) + + # 本地硬盘录像 video_writer.write(current_frame) - # 计算并渲染接收端的显示 FPS + # 计算并渲染本地显示的实时 FPS fps_counter += 1 if (time.time() - fps_start_time) > 1.0: fps_display = fps_counter fps_counter = 0 fps_start_time = time.time() - # 在右上角标注实时的图传显示帧率 + # 标注并进行本地窗口渲染显示 cv2.putText(current_frame, f"GCS FPS: {fps_display}", (current_frame.shape[1] - 140, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) - - # 实时监控显示 cv2.imshow("🚀 UAV AI Ground Station Monitor", current_frame) - # 按键检测（按 'q' 安全退出） if cv2.waitKey(1) & 0xFF == ord('q'): print("👋 接收到关闭指令...") break - # 严格控制写入频率，使其匹配 24.0 FPS 的时间流速 + # 严格步长控制 time_used = time.time() - loop_start sleep_time = frame_interval - time_used if sleep_time > 0: time.sleep(sleep_time) finally: - # 善后处理：确保内存日志和视频安全落盘 is_running = False video_writer.release() cv2.destroyAllWindows() - print("\n=======================================================") - print(f"✅ 地面站已安全关闭。") - print(f"🎬 带有精确检测框、对准线和控制误差的视频已成功保存在:\n {SAVE_PATH}") - print("=======================================================") + proxy.stop() # 安全关闭转码流容器 + print(f"\n✅ 地面站双轨服务已安全关闭。录像保存在: {SAVE_PATH}") if __name__ == "__main__":