JPEG转换为H.264(集成在接收脚本中)

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=======================================================")
+        proxy.stop()  # 安全关闭转码流容器
-        print(f"✅ 地面站已安全关闭。")
+        print(f"\n✅ 地面站双轨服务已安全关闭。录像保存在: {SAVE_PATH}")
-        print(f"🎬 带有精确检测框、对准线和控制误差的视频已成功保存在:\n   {SAVE_PATH}")
-        print("=======================================================")
 
 
 if __name__ == "__main__":