添加使用ffmpeg推流功能

config.yaml

@@ -50,4 +50,18 @@ logging:
 
 # 中文字体设置
 display:
-  font_path: "/usr/share/fonts/truetype/noto/NotoMono-Regular.ttf"
+  font_path: "/usr/share/fonts/truetype/noto/NotoMono-Regular.ttf"
+
+
+stream:
+  enabled: true  # 开关推流功能
+  rtmp_url: "rtsp://127.0.0.1/live/video6"
+  
+  ffmpeg:
+    fps: 25  # 推流帧率
+    video_bitrate: "2000k"  # 码率
+    preset: "ultrafast"  # 编码速度
+    tune: "zerolatency"  # 低延迟优化
+    pixel_format: "yuv420p"
+    video_codec: "libx264"
+    audio: false  # 不包含音频

face_rec.py

@@ -13,6 +13,8 @@ from compreface.service import RecognitionService, DetectionService
 import time
 from collections import defaultdict
 from PIL import Image, ImageDraw, ImageFont
+import subprocess
+import sys
 
 
 class FaceRecognitionSystem:
@@ -73,6 +75,14 @@ class FaceRecognitionSystem:
         self.font_medium = ImageFont.truetype(self.font_path, 24)
         self.font_large = ImageFont.truetype(self.font_path, 28)
         
+        # FFmpeg推流进程
+        self.ffmpeg_process = None
+        self.stream_enabled = self.config.get('stream', {}).get('enabled', False)
+        self.stream_retry_count = 0  # 推流重试计数
+        self.stream_max_retries = 5  # 最大重试次数
+        self.stream_last_retry_time = None  # 上次重试时间
+        self.stream_retry_cooldown = 10  # 重试冷却时间(秒)
+        
         self.logger.info("人脸识别系统初始化完成")
     
     def _setup_logging(self):
@@ -126,10 +136,8 @@ class FaceRecognitionSystem:
         # Linux系统
         else:
             font_paths = [
-                # "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf",
                 "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
                 "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc",
-                # "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
             ]
         
         # 查找可用的字体
@@ -149,6 +157,190 @@ class FaceRecognitionSystem:
         self.logger.warning("未找到中文字体,将使用默认字体(可能无法显示中文)")
         return "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"
     
+    def _init_ffmpeg_stream(self):
+        """初始化FFmpeg推流"""
+        if not self.stream_enabled:
+            self.logger.info("流媒体推流功能未启用")
+            return False
+        
+        # 检查是否在冷却期内
+        if self.stream_last_retry_time:
+            elapsed = time.time() - self.stream_last_retry_time
+            if elapsed < self.stream_retry_cooldown:
+                self.logger.debug(f"推流重试冷却中,还需等待 {self.stream_retry_cooldown - elapsed:.1f} 秒")
+                return False
+        
+        # # 检查是否超过最大重试次数
+        # if self.stream_retry_count >= self.stream_max_retries:
+        #     self.logger.error(f"推流失败次数已达到上限({self.stream_max_retries}次),推流功能已禁用")
+        #     self.logger.error("请检查ZLMediaKit是否正常运行,以及推流地址是否正确")
+        #     self.stream_enabled = False
+        #     return False
+        
+        stream_config = self.config['stream']
+        ffmpeg_config = stream_config['ffmpeg']
+        stream_url = stream_config.get('rtmp_url') or stream_config.get('stream_url', '')
+        
+        # 获取摄像头分辨率
+        cam_config = self.config['camera']
+        # 计算带状态面板的总高度
+        panel_height = 200
+        total_height = cam_config['height'] + panel_height
+        
+        # 根据推流URL判断输出格式
+        if stream_url.startswith('rtmp://'):
+            output_format = 'flv'
+            self.logger.debug("使用RTMP协议推流")
+        elif stream_url.startswith('rtsp://'):
+            output_format = 'rtsp'
+            self.logger.debug("使用RTSP协议推流")
+        else:
+            self.logger.error(f"不支持的推流协议: {stream_url}")
+            return False
+        
+        # 构建FFmpeg命令
+        ffmpeg_cmd = [
+            'ffmpeg',
+            '-y',  # 覆盖输出文件
+            '-f', 'rawvideo',  # 输入格式
+            # '-vcodec', 'rawvideo',
+            '-pix_fmt', 'bgr24',  # OpenCV使用BGR格式
+            '-s', f"{cam_config['width']}x{total_height}",  # 输入分辨率
+            '-r', str(ffmpeg_config['fps']),  # 输入帧率
+            '-i', '-',  # 从stdin读取
+            '-c:v', ffmpeg_config['video_codec'],  # 视频编码器
+            '-pix_fmt', ffmpeg_config['pixel_format'],  # 输出像素格式
+            # '-preset', ffmpeg_config['preset'],  # 编码预设
+            '-tune', ffmpeg_config['tune'],  # 编码调优
+            '-b:v', ffmpeg_config['video_bitrate'],  # 视频码率
+            # '-r', str(ffmpeg_config['fps']),  # 输出帧率
+            '-g', str(ffmpeg_config['fps'] * 2),  # GOP大小(2秒)
+        ]
+        
+        # 如果不需要音频,添加-an参数
+        if not ffmpeg_config.get('audio', False):
+            ffmpeg_cmd.append('-an')
+        
+        # 添加RTMP特定参数
+        if output_format == 'flv':
+            ffmpeg_cmd.extend([
+                '-f', 'flv',
+                '-flvflags', 'no_duration_filesize',  # 直播流不需要duration
+                stream_url
+            ])
+        # 添加RTSP特定参数
+        elif output_format == 'rtsp':
+            ffmpeg_cmd.extend([
+                '-f', 'rtsp',
+                '-rtsp_transport', 'tcp',  # 使用TCP传输
+                stream_url
+            ])
+        
+        try:
+            self.stream_retry_count += 1
+            self.stream_last_retry_time = time.time()
+            
+            self.logger.info(f"启动FFmpeg推流 (第{self.stream_retry_count}次尝试): {stream_url}")
+            self.logger.debug(f"FFmpeg命令: {' '.join(ffmpeg_cmd)}")
+            
+            # 启动FFmpeg进程
+            self.ffmpeg_process = subprocess.Popen(
+                ffmpeg_cmd,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,  # 忽略标准输出
+                stderr=subprocess.PIPE,
+                bufsize=10**8
+            )
+            
+            # 等待一小段时间检查进程是否立即失败
+            time.sleep(0.5)
+            if self.ffmpeg_process.poll() is not None:
+                # 进程已退出,读取错误信息
+                stderr_output = self.ffmpeg_process.stderr.read().decode('utf-8', errors='ignore')
+                self.logger.error(f"FFmpeg进程启动失败:")
+                # 只显示关键错误信息
+                for line in stderr_output.split('\n'):
+                    if 'error' in line.lower() or 'failed' in line.lower() or 'connection' in line.lower():
+                        self.logger.error(f"  {line.strip()}")
+                self.ffmpeg_process = None
+                return False
+            
+            self.logger.info("FFmpeg推流进程启动成功")
+            # 重置重试计数(启动成功)
+            self.stream_retry_count = 0
+            return True
+            
+        except FileNotFoundError:
+            self.logger.error("FFmpeg未安装或不在系统PATH中,推流功能将被禁用")
+            self.stream_enabled = False
+            self.ffmpeg_process = None
+            return False
+        except Exception as e:
+            self.logger.error(f"启动FFmpeg推流失败: {e}")
+            self.ffmpeg_process = None
+            return False
+    
+    def _push_frame_to_stream(self, frame: np.ndarray):
+        """推送帧到流媒体服务器"""
+        if not self.stream_enabled or self.ffmpeg_process is None:
+            return
+        
+        try:
+            # 检查FFmpeg进程是否仍在运行
+            if self.ffmpeg_process.poll() is not None:
+                # 进程已退出,读取错误信息
+                if self.stream_retry_count < self.stream_max_retries:
+                    stderr_output = self.ffmpeg_process.stderr.read().decode('utf-8', errors='ignore')
+                    if stderr_output:
+                        self.logger.error(f"FFmpeg进程异常退出: {stderr_output[:300]}")
+                    else:
+                        self.logger.warning("FFmpeg进程已退出")
+                return
+            
+            # 写入帧数据到FFmpeg的stdin
+            self.ffmpeg_process.stdin.write(frame.tobytes())
+            self.ffmpeg_process.stdin.flush()  # 立即刷新缓冲区
+            
+        except BrokenPipeError:
+            if self.stream_retry_count < self.stream_max_retries:
+                self.logger.warning("FFmpeg推流管道断开")
+            self.ffmpeg_process = None
+        except Exception as e:
+            if self.stream_retry_count < self.stream_max_retries:
+                self.logger.error(f"推送帧到流媒体失败: {e}")
+            self.ffmpeg_process = None
+    
+    def _close_ffmpeg_stream(self):
+        """关闭FFmpeg推流"""
+        if self.ffmpeg_process:
+            try:
+                self.logger.info("正在关闭FFmpeg推流...")
+                
+                # 先关闭stdin
+                if self.ffmpeg_process.stdin and not self.ffmpeg_process.stdin.closed:
+                    try:
+                        self.ffmpeg_process.stdin.close()
+                    except:
+                        pass
+                
+                # 等待进程结束
+                try:
+                    self.ffmpeg_process.wait(timeout=3)
+                    self.logger.info("FFmpeg推流已关闭")
+                except subprocess.TimeoutExpired:
+                    self.logger.warning("FFmpeg进程未响应,强制终止")
+                    self.ffmpeg_process.kill()
+                    self.ffmpeg_process.wait()
+                    
+            except Exception as e:
+                self.logger.debug(f"关闭FFmpeg推流时出现异常: {e}")
+                try:
+                    self.ffmpeg_process.kill()
+                except:
+                    pass
+            finally:
+                self.ffmpeg_process = None
+    
     def _init_compreface(self):
         """初始化CompreFace SDK"""
         cf_config = self.config['compreface']
@@ -575,6 +767,10 @@ class FaceRecognitionSystem:
         """处理视频流"""
         self._init_camera()
         
+        # 初始化FFmpeg推流
+        if self.stream_enabled:
+            self._init_ffmpeg_stream()
+        
         frame_interval = self.config['face_detection']['frame_interval']
         quality_threshold = self.config['face_detection']['quality_threshold']
         face_duration = self.config['face_detection']['face_present_duration']
@@ -693,7 +889,6 @@ class FaceRecognitionSystem:
                                         }
                                     }
                                     await self.send_websocket_message(reception_msg)
-                                    self.logger.info(f"发送消息: {reception_msg}")
                                     
                                     # 记录识别时间
                                     self.recognition_history[person_id] = datetime.now()
@@ -745,6 +940,20 @@ class FaceRecognitionSystem:
                 display_frame = self.draw_info_on_frame(frame)
                 cv2.imshow(window_name, display_frame)
                 
+                # 推送到流媒体服务器
+                if self.stream_enabled:
+                    # 如果FFmpeg进程不存在或已退出,尝试重新启动
+                    if self.ffmpeg_process is None or self.ffmpeg_process.poll() is not None:
+                        if self.stream_retry_count < self.stream_max_retries:
+                            success = self._init_ffmpeg_stream()
+                            if not success:
+                                # 启动失败,等待冷却期
+                                pass
+                    
+                    # 推送帧
+                    if self.ffmpeg_process and self.ffmpeg_process.poll() is None:
+                        self._push_frame_to_stream(display_frame)
+                
                 # 检查键盘输入
                 key = cv2.waitKey(1) & 0xFF
                 if key == ord('q') or key == 27:  # 'q' 或 ESC 退出
@@ -756,9 +965,15 @@ class FaceRecognitionSystem:
         except Exception as e:
             self.logger.error(f"视频流处理错误: {e}")
         finally:
+            # 清理资源
             if self.camera:
                 self.camera.release()
                 self.logger.info("摄像头已释放")
+            
+            # 关闭FFmpeg推流
+            if self.stream_enabled:
+                self._close_ffmpeg_stream()
+            
             cv2.destroyAllWindows()
     
     async def run(self):