diff --git a/pipeline/__init__.py b/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pipeline/pipeline.py b/pipeline/pipeline.py index 1789986..c4243df 100644 --- a/pipeline/pipeline.py +++ b/pipeline/pipeline.py @@ -6,6 +6,7 @@ from .pipeline_data import PipelineData class Pipeline: """管道核心类,管理多个处理器,负责数据流转""" + def __init__(self): self.processors: List[BaseProcessor] = [] # 同步处理器列表 self.async_processors: List[BaseProcessor] = [] # 异步处理器列表 @@ -43,6 +44,7 @@ class Pipeline: :param data_iterator: 数据源迭代器(如视频+YOLO的迭代器) :return: 处理后的数据包迭代器 """ + # 创建异步任务列表,用于管理异步处理器 async def run_async_processors(data: PipelineData): tasks = [] @@ -55,8 +57,12 @@ class Pipeline: for data in data_iterator: # 依次执行每个同步处理器的处理逻辑 for processor in self.processors: - data = processor.process(data) - + try: + data = processor.process(data) + yield data + except Exception as e: + print(f"同步处理器 {processor.name} 执行出错: {e}") + # 异步执行异步处理器,不阻塞数据流 if self.async_processors: # 在独立的事件循环中运行异步处理器(如果可用) @@ -68,7 +74,7 @@ class Pipeline: except RuntimeError: # 如果没有运行中的事件循环,则创建一个新的 asyncio.run(run_async_processors(data.copy() if hasattr(data, 'copy') else data)) - + yield data # 返回处理后的数据包 async def _process_async(self, processor, data): @@ -81,4 +87,4 @@ class Pipeline: # 如果处理器不支持异步处理,则同步处理 processor.process(data) except Exception as e: - print(f"异步处理器 {processor.name} 执行出错: {e}") \ No newline at end of file + print(f"异步处理器 {processor.name} 执行出错: {e}")