处理上下文 - 在整个处理流程中共享的数据 每个模块可以从context中读取需要的数据,也可以往里写入处理结果。 这样不同模块之间可以通过context传递数据。
| 22 | |
| 23 | @dataclass |
| 24 | class ProcessingContext: |
| 25 | """ |
| 26 | 处理上下文 - 在整个处理流程中共享的数据 |
| 27 | |
| 28 | 每个模块可以从context中读取需要的数据,也可以往里写入处理结果。 |
| 29 | 这样不同模块之间可以通过context传递数据。 |
| 30 | """ |
| 31 | # === 输入信息 === |
| 32 | image_path: str # 原始图片路径 |
| 33 | canvas_width: int = 0 # 画布宽度 |
| 34 | canvas_height: int = 0 # 画布高度 |
| 35 | |
| 36 | # === 元素数据(各模块共享) === |
| 37 | elements: List['ElementInfo'] = field(default_factory=list) |
| 38 | |
| 39 | # === XML片段(各子模块生成,XMLMerger合并) === |
| 40 | xml_fragments: List['XMLFragment'] = field(default_factory=list) |
| 41 | |
| 42 | # === 配置 === |
| 43 | config: Optional['ProcessingConfig'] = None |
| 44 | output_dir: str = "./output" |
| 45 | |
| 46 | # === 模型实例(可共享,避免重复加载) === |
| 47 | shared_models: Dict[str, Any] = field(default_factory=dict) |
| 48 | |
| 49 | # === 中间结果(用于debug或二次处理) === |
| 50 | intermediate_results: Dict[str, Any] = field(default_factory=dict) |
| 51 | |
| 52 | def get_elements_by_type(self, element_type: str) -> List['ElementInfo']: |
| 53 | """按类型获取元素""" |
| 54 | return [e for e in self.elements if e.element_type == element_type] |
| 55 | |
| 56 | def get_elements_without_xml(self) -> List['ElementInfo']: |
| 57 | """获取还没有生成XML的元素""" |
| 58 | return [e for e in self.elements if not e.has_xml()] |
| 59 | |
| 60 | def get_elements_with_xml(self) -> List['ElementInfo']: |
| 61 | """获取已经生成XML的元素""" |
| 62 | return [e for e in self.elements if e.has_xml()] |
| 63 | |
| 64 | def add_xml_fragment(self, fragment: 'XMLFragment'): |
| 65 | """添加XML片段""" |
| 66 | self.xml_fragments.append(fragment) |
| 67 | |
| 68 | |
| 69 | class BaseProcessor(ABC): |
no outgoing calls
no test coverage detected