fix(pipeline): 移除同步处理器中的错误yield语句

- 解决流中异常返回

app.py

@@ -1,4 +1,5 @@
 import asyncio
+from loguru import logger
 
 from yolo_gs.pipeline.handler.DrawDirectionProcessor import DrawDirectionProcessor
 from yolo_gs.pipeline.handler.DrawGraffitiProcessor import DrawGraffitiProcessor
@@ -12,8 +13,23 @@ import cv2
 from yolo_gs.pipeline.data_source import video_yolo_iterator
 from yolo_gs.pipeline.pipeline import Pipeline
 
+
+def init_logger():
+    logger.add(
+        "logs/yolo_gs.log",
+        rotation="100 MB",  # 原有配置：日志文件达100MB轮转
+        level="DEBUG",  # 原有配置：记录DEBUG及以上级别
+        enqueue=True,  # 开启异步输出（关键参数）
+        backtrace=True,  # 记录完整异常堆栈（包括第三方库）
+        diagnose=True,  # 显示异常时的变量值（便于调试，生产环境可设为False）
+    )
+
+
 # ===================== 测试运行 =====================
 if __name__ == "__main__":
+    init_logger()
+    logger.info("开始运行YOLO涂鸦车道检测系统")
+
     # 1. 创建管道
     pipeline = Pipeline()
 
@@ -25,12 +41,12 @@ if __name__ == "__main__":
     pipeline.add_processor(DrawObjectBoxProcessor())
     pipeline.add_processor(ResultLogger(name="结果日志器"))
 
-
     # 4. 构建视频+YOLO跟踪迭代器（替换为你的视频路径，或0使用摄像头）
     video_path = "123.mp4"  # 或 0 (摄像头)
     try:
+        logger.info(f"开始处理视频: {video_path}")
         # 使用跟踪迭代器，可以获取对象ID
-        data_iter = video_yolo_iterator(video_path, model_path="yolo11s.pt",tracker="botsort.yaml", device="0")
+        data_iter = video_yolo_iterator(video_path, model_path="yolo11s.pt", tracker="botsort.yaml", device="0")
 
         # 5. 运行管道，迭代处理数据
         for processed_data in pipeline.run(data_iter):
@@ -40,8 +56,10 @@ if __name__ == "__main__":
                 # 直接显示经过管道处理后的帧，其中已经包含了绘制处理器的效果
                 cv2.imshow("YOLO Tracking Pipeline", frame)
                 if cv2.waitKey(1) & 0xFF == ord('q'):
+                    logger.info("用户请求退出程序")
                     break
     except Exception as e:
-        print(f"运行出错: {e}")
+        logger.error(f"运行出错: {e}")
     finally:
-        cv2.destroyAllWindows()
+        cv2.destroyAllWindows()
+        logger.info("程序结束运行")

pipeline/base_processor.py

@@ -1,3 +1,4 @@
+from loguru import logger
 from abc import ABC, abstractmethod
 import asyncio
 from .pipeline_data import PipelineData
@@ -7,6 +8,9 @@ class BaseProcessor(ABC):
     """所有处理器的抽象基类，定义统一的处理接口"""
     def __init__(self, name: str):
         self.name = name  # 处理器名称（便于日志/调试）
+        # 初始化日志记录器
+        self.logger = logger
+        self.logger.debug(f"处理器 {self.name} 已初始化")
 
     @abstractmethod
     def process(self, data: PipelineData) -> PipelineData:

pipeline/data_source.py

@@ -1,3 +1,4 @@
+from loguru import logger
 from typing import Iterator
 import cv2
 from ultralytics import YOLO
@@ -16,15 +17,18 @@ def video_yolo_iterator(video_path: str, model_path: str = "yolov8n.pt", tracker
     """
     # 加载YOLO模型
     model = YOLO(model_path)
+    logger.info(f"已加载YOLO模型: {model_path}")
     # 打开视频
     cap = cv2.VideoCapture(video_path)
     if not cap.isOpened():
+        logger.error(f"无法打开视频: {video_path}")
         raise ValueError(f"无法打开视频: {video_path}")
 
     frame_idx = 0
     try:
         # 使用 model.track 替代 model 进行跟踪检测
         results_stream = model.track(source=video_path, tracker=tracker, device=device, stream=True)
+        logger.info(f"已开始跟踪视频: {video_path}")
 
         last_data = PipelineData()
         for r in results_stream:  # 迭代跟踪结果
@@ -67,6 +71,7 @@ def video_yolo_detect_iterator(video_path: str, model_path: str = "yolov8n.pt",
     # 打开视频
     cap = cv2.VideoCapture(video_path)
     if not cap.isOpened():
+        logger.error(f"无法打开视频: {video_path}")
         raise ValueError(f"无法打开视频: {video_path}")
 
     frame_idx = 0
@@ -94,4 +99,4 @@ def video_yolo_detect_iterator(video_path: str, model_path: str = "yolov8n.pt",
                 frame_idx += 1
     finally:
         # 确保视频流关闭
-        cap.release()
+        cap.release()

pipeline/handler/DrawDirectionProcessor.py

@@ -2,8 +2,8 @@ import cv2
 import numpy as np
 import math
 from collections import deque
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 
 
 class DrawDirectionProcessor(BaseProcessor):
@@ -96,7 +96,7 @@ class DrawDirectionProcessor(BaseProcessor):
                 self._draw_track_info(vis, track_id, (center_x, center_y))
 
             except Exception as e:
-                print(f"Error processing box: {e}")
+                self.logger.error(f"Error processing box: {e}")
                 continue
 
         # 清理长时间未出现的track

pipeline/handler/DrawGraffitiProcessor.py

@@ -1,8 +1,8 @@
 import cv2
 import numpy as np
 import math
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 
 
 class DrawGraffitiProcessor(BaseProcessor):

pipeline/handler/DrawObjectBoxProcessor.py

@@ -2,8 +2,8 @@ import cv2
 import numpy as np
 from dataclasses import dataclass, field
 from typing import List, Optional, Set
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 
 
 @dataclass

pipeline/handler/GraffitiProcessor.py

@@ -2,8 +2,8 @@ import numpy as np
 import cv2
 import math
 from collections import deque, defaultdict
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 
 
 class GraffitiProcessor(BaseProcessor):

pipeline/handler/ResultLogger.py

@@ -5,6 +5,6 @@ import numpy as np
 class ResultLogger(BaseProcessor):
     """示例处理器：打印检测结果日志"""
     def process(self, data: PipelineData) -> PipelineData:
-        print(f"\n【{self.name}】帧{data.frame_idx} - 检测到目标数: {len(data.current_result.boxes)}")
+        self.logger.info(f"【{self.name}】帧{data.frame_idx} - 检测到目标数: {len(data.current_result.boxes)}")
-        print(f"缓存帧数: {len(data.result_cache)}")
+        self.logger.info(f"缓存帧数: {len(data.result_cache)}")
         return data

pipeline/handler/RetrogradeProcessor.py

@@ -1,7 +1,7 @@
 import cv2
 import numpy as np
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 import math
 
 
@@ -115,7 +115,7 @@ class RetrogradeProcessor(BaseProcessor):
 
         # 输出逆行信息
         for track_id, event in events_all0.items():
-            print(f"【逆行检测器】帧{self.frame_idx} - 检测到事件：{event} - 轨迹ID: {track_id}")
+            self.logger.info(f"【逆行检测器】帧{self.frame_idx} - 检测到事件：{event} - 轨迹ID: {track_id}")
 
         # 更新事件历史
         for track_id, event in frame_events.items():

pipeline/handler/graffiti_visualizer.py

@@ -1,8 +1,8 @@
 import numpy as np
 import cv2
 import math
-from yolo_gs.pipeline.base_processor import BaseProcessor
+from ..base_processor import BaseProcessor
-from yolo_gs.pipeline.pipeline_data import PipelineData
+from ..pipeline_data import PipelineData
 
 
 class GraffitiVisualizer(BaseProcessor):

pipeline/pipeline.py

@@ -1,3 +1,4 @@
+from loguru import logger
 from typing import List, Iterator
 import asyncio
 from .base_processor import BaseProcessor
@@ -10,16 +11,19 @@ class Pipeline:
     def __init__(self):
         self.processors: List[BaseProcessor] = []  # 同步处理器列表
         self.async_processors: List[BaseProcessor] = []  # 异步处理器列表
+        # 初始化日志记录器
+        self.logger = logger
+        self.logger.debug("管道已初始化")
 
     def add_processor(self, processor: BaseProcessor) -> None:
         """向管道添加同步处理器（按添加顺序执行）"""
         self.processors.append(processor)
-        print(f"管道已添加同步处理器: {processor}")
+        self.logger.info(f"管道已添加同步处理器: {processor}")
 
     def add_processor_async(self, processor: BaseProcessor) -> None:
         """向管道添加异步处理器（异步执行，不阻塞数据流）"""
         self.async_processors.append(processor)
-        print(f"管道已添加异步处理器: {processor}")
+        self.logger.info(f"管道已添加异步处理器: {processor}")
 
     def remove_processor(self, processor_name: str) -> bool:
         """移除指定名称的处理器"""
@@ -27,15 +31,15 @@ class Pipeline:
         for idx, p in enumerate(self.processors):
             if p.name == processor_name:
                 self.processors.pop(idx)
-                print(f"管道已移除同步处理器: {p}")
+                self.logger.info(f"管道已移除同步处理器: {p}")
                 return True
         # 检查异步处理器列表
         for idx, p in enumerate(self.async_processors):
             if p.name == processor_name:
                 self.async_processors.pop(idx)
-                print(f"管道已移除异步处理器: {p}")
+                self.logger.info(f"管道已移除异步处理器: {p}")
                 return True
-        print(f"未找到处理器: {processor_name}")
+        self.logger.warning(f"未找到处理器: {processor_name}")
         return False
 
     def run(self, data_iterator: Iterator[PipelineData]) -> Iterator[PipelineData]:
@@ -59,9 +63,8 @@ class Pipeline:
             for processor in self.processors:
                 try:
                     data = processor.process(data)
-                    yield data
                 except Exception as e:
-                    print(f"同步处理器 {processor.name} 执行出错: {e}")
+                    self.logger.error(f"同步处理器 {processor.name} 执行出错: {e}")
 
             # 异步执行异步处理器，不阻塞数据流
             if self.async_processors:
@@ -87,4 +90,4 @@ class Pipeline:
                 # 如果处理器不支持异步处理，则同步处理
                 processor.process(data)
         except Exception as e:
-            print(f"异步处理器 {processor.name} 执行出错: {e}")
+            self.logger.error(f"异步处理器 {processor.name} 执行出错: {e}")

requirements.txt

@@ -1 +1,2 @@
-ultralytics>=8.3.249
+ultralytics>=8.3.249
+loguru>=0.7.3