pengxiao
219232de74
以讲义内容为骨架迁移到标准目录格式： - officefile/ 主内容（12章 + 附录 + CC4SI补充） - dofile/ 代码示例（11个Python脚本） - data/ 图片资源 - output/ 生成输出（忽略） - Archive/ 归档旧目录（忽略） - .claude/skills/ 保留markdown-to-docx工具链 - .pandoc/ 保留CSL和本地化配置 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
685 lines
20 KiB
Python
685 lines
20 KiB
Python
"""
|
|
模块化系统示例 (Modular System Example)
|
|
========================================
|
|
|
|
本示例展示了空间智能系统的模块化设计原则。
|
|
模块化是构建可维护、可扩展系统的基础。
|
|
|
|
核心概念:
|
|
1. 关注点分离 - 每个模块负责特定功能
|
|
2. 接口设计 - 定义清晰的模块间通信协议
|
|
3. 依赖注入 - 降低模块间耦合
|
|
4. 插件架构 - 支持动态扩展功能
|
|
|
|
作者: CC4SI 项目组
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, List, Any, Optional, Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
import json
|
|
|
|
|
|
# ============================================================================
|
|
# 模块接口定义 (抽象基类)
|
|
# ============================================================================
|
|
|
|
class ModuleType(Enum):
|
|
"""模块类型枚举"""
|
|
DATA_LOADER = "data_loader"
|
|
DATA_PROCESSOR = "data_processor"
|
|
ANALYZER = "analyzer"
|
|
VISUALIZER = "visualizer"
|
|
EXPORTER = "exporter"
|
|
|
|
|
|
class ModuleStatus(Enum):
|
|
"""模块状态枚举"""
|
|
IDLE = "idle"
|
|
INITIALIZING = "initializing"
|
|
READY = "ready"
|
|
RUNNING = "running"
|
|
ERROR = "error"
|
|
|
|
|
|
@dataclass
|
|
class ModuleMetadata:
|
|
"""模块元数据"""
|
|
name: str
|
|
version: str
|
|
module_type: ModuleType
|
|
description: str = ""
|
|
dependencies: List[str] = field(default_factory=list)
|
|
author: str = ""
|
|
config_schema: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class IModule(ABC):
|
|
"""
|
|
模块接口 - 所有模块必须实现此接口
|
|
|
|
这是一个抽象基类，定义了所有模块必须遵循的契约。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
"""
|
|
初始化模块
|
|
|
|
Args:
|
|
config: 模块配置字典
|
|
"""
|
|
self.config = config or {}
|
|
self.status = ModuleStatus.IDLE
|
|
self._context: Optional['ModuleContext'] = None
|
|
|
|
@property
|
|
@abstractmethod
|
|
def metadata(self) -> ModuleMetadata:
|
|
"""返回模块元数据"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def initialize(self, context: 'ModuleContext') -> bool:
|
|
"""
|
|
初始化模块
|
|
|
|
Args:
|
|
context: 模块上下文，提供对系统资源的访问
|
|
|
|
Returns:
|
|
初始化是否成功
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def execute(self, input_data: Any) -> Any:
|
|
"""
|
|
执行模块功能
|
|
|
|
Args:
|
|
input_data: 输入数据
|
|
|
|
Returns:
|
|
处理结果
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def shutdown(self) -> None:
|
|
"""关闭模块，释放资源"""
|
|
pass
|
|
|
|
def get_config(self, key: str, default: Any = None) -> Any:
|
|
"""获取配置值"""
|
|
return self.config.get(key, default)
|
|
|
|
def set_config(self, key: str, value: Any) -> None:
|
|
"""设置配置值"""
|
|
self.config[key] = value
|
|
|
|
|
|
# ============================================================================
|
|
# 模块上下文 - 提供模块间通信
|
|
# ============================================================================
|
|
|
|
class ModuleContext:
|
|
"""
|
|
模块上下文
|
|
|
|
提供模块间通信和资源共享机制，实现松耦合设计。
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._modules: Dict[str, IModule] = {}
|
|
self._shared_data: Dict[str, Any] = {}
|
|
self._event_handlers: Dict[str, List[Callable]] = {}
|
|
|
|
def register_module(self, name: str, module: IModule) -> bool:
|
|
"""注册模块"""
|
|
if name in self._modules:
|
|
print(f"警告: 模块 '{name}' 已存在，将被覆盖")
|
|
self._modules[name] = module
|
|
print(f"模块 '{name}' 已注册 (类型: {module.metadata.module_type.value})")
|
|
return True
|
|
|
|
def get_module(self, name: str) -> Optional[IModule]:
|
|
"""获取模块实例"""
|
|
return self._modules.get(name)
|
|
|
|
def has_module(self, name: str) -> bool:
|
|
"""检查模块是否存在"""
|
|
return name in self._modules
|
|
|
|
def set_shared_data(self, key: str, value: Any) -> None:
|
|
"""设置共享数据"""
|
|
self._shared_data[key] = value
|
|
|
|
def get_shared_data(self, key: str, default: Any = None) -> Any:
|
|
"""获取共享数据"""
|
|
return self._shared_data.get(key, default)
|
|
|
|
def subscribe_event(self, event_name: str, handler: Callable) -> None:
|
|
"""订阅事件"""
|
|
if event_name not in self._event_handlers:
|
|
self._event_handlers[event_name] = []
|
|
self._event_handlers[event_name].append(handler)
|
|
|
|
def publish_event(self, event_name: str, *args, **kwargs) -> None:
|
|
"""发布事件"""
|
|
if event_name in self._event_handlers:
|
|
for handler in self._event_handlers[event_name]:
|
|
handler(*args, **kwargs)
|
|
|
|
|
|
# ============================================================================
|
|
# 模块基类 - 提供通用功能实现
|
|
# ============================================================================
|
|
|
|
class BaseModule(IModule):
|
|
"""
|
|
模块基类
|
|
|
|
提供IModule接口的默认实现，子类只需实现特定功能。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
super().__init__(config)
|
|
self._metadata: Optional[ModuleMetadata] = None
|
|
|
|
@property
|
|
def metadata(self) -> ModuleMetadata:
|
|
if self._metadata is None:
|
|
raise NotImplementedError("子类必须设置 _metadata")
|
|
return self._metadata
|
|
|
|
def initialize(self, context: ModuleContext) -> bool:
|
|
"""默认初始化实现"""
|
|
self._context = context
|
|
self.status = ModuleStatus.INITIALIZING
|
|
|
|
# 检查依赖
|
|
for dep in self.metadata.dependencies:
|
|
if not context.has_module(dep):
|
|
print(f"错误: 依赖模块 '{dep}' 不存在")
|
|
self.status = ModuleStatus.ERROR
|
|
return False
|
|
|
|
self.status = ModuleStatus.READY
|
|
print(f"模块 '{self.metadata.name}' 初始化完成")
|
|
return True
|
|
|
|
def shutdown(self) -> None:
|
|
"""默认关闭实现"""
|
|
self.status = ModuleStatus.IDLE
|
|
print(f"模块 '{self.metadata.name}' 已关闭")
|
|
|
|
|
|
# ============================================================================
|
|
# 具体模块实现
|
|
# ============================================================================
|
|
|
|
class CSVDataLoaderModule(BaseModule):
|
|
"""
|
|
CSV 数据加载模块
|
|
|
|
负责从CSV文件加载空间数据。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
super().__init__(config)
|
|
self._metadata = ModuleMetadata(
|
|
name="csv_data_loader",
|
|
version="1.0.0",
|
|
module_type=ModuleType.DATA_LOADER,
|
|
description="从CSV文件加载空间数据",
|
|
author="CC4SI"
|
|
)
|
|
self._data: List[Dict[str, Any]] = []
|
|
|
|
def execute(self, input_data: Any) -> Any:
|
|
"""
|
|
执行数据加载
|
|
|
|
Args:
|
|
input_data: 文件路径或模拟数据
|
|
|
|
Returns:
|
|
加载的数据列表
|
|
"""
|
|
self.status = ModuleStatus.RUNNING
|
|
|
|
if isinstance(input_data, str):
|
|
# 实际场景中应从文件读取
|
|
print(f"从文件 '{input_data}' 加载数据...")
|
|
# 模拟加载
|
|
self._data = self._load_sample_data()
|
|
elif isinstance(input_data, list):
|
|
self._data = input_data
|
|
else:
|
|
self._data = self._load_sample_data()
|
|
|
|
# 将数据存入共享上下文
|
|
if self._context:
|
|
self._context.set_shared_data("raw_data", self._data)
|
|
self._context.publish_event("data_loaded", len(self._data))
|
|
|
|
self.status = ModuleStatus.READY
|
|
return self._data
|
|
|
|
def _load_sample_data(self) -> List[Dict[str, Any]]:
|
|
"""加载示例数据"""
|
|
return [
|
|
{"id": 1, "x": 10, "y": 20, "value": 100, "type": "A"},
|
|
{"id": 2, "x": 30, "y": 40, "value": 200, "type": "B"},
|
|
{"id": 3, "x": 50, "y": 60, "value": 150, "type": "A"},
|
|
{"id": 4, "x": 70, "y": 80, "value": 300, "type": "C"},
|
|
{"id": 5, "x": 90, "y": 100, "value": 250, "type": "B"},
|
|
]
|
|
|
|
|
|
class DataValidationModule(BaseModule):
|
|
"""
|
|
数据验证模块
|
|
|
|
负责验证数据质量和完整性。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
super().__init__(config)
|
|
self._metadata = ModuleMetadata(
|
|
name="data_validator",
|
|
version="1.0.0",
|
|
module_type=ModuleType.DATA_PROCESSOR,
|
|
description="验证数据质量和完整性",
|
|
dependencies=["csv_data_loader"],
|
|
author="CC4SI"
|
|
)
|
|
self.validation_rules: List[Callable] = []
|
|
|
|
def add_validation_rule(self, rule: Callable[[Dict], bool], name: str = ""):
|
|
"""添加验证规则"""
|
|
self.validation_rules.append(rule)
|
|
if name:
|
|
print(f"添加验证规则: {name}")
|
|
|
|
def execute(self, input_data: Any) -> Any:
|
|
"""
|
|
执行数据验证
|
|
|
|
Args:
|
|
input_data: 待验证的数据
|
|
|
|
Returns:
|
|
验证结果
|
|
"""
|
|
self.status = ModuleStatus.RUNNING
|
|
|
|
if not isinstance(input_data, list):
|
|
return {"valid": False, "errors": ["输入数据格式错误"]}
|
|
|
|
errors = []
|
|
warnings = []
|
|
|
|
for i, item in enumerate(input_data):
|
|
# 检查必需字段
|
|
if "id" not in item:
|
|
errors.append(f"第 {i} 项缺少 'id' 字段")
|
|
if "x" not in item or "y" not in item:
|
|
errors.append(f"第 {i} 项缺少坐标字段")
|
|
|
|
# 应用自定义验证规则
|
|
for rule in self.validation_rules:
|
|
try:
|
|
if not rule(item):
|
|
warnings.append(f"第 {i} 项未通过自定义规则验证")
|
|
except Exception as e:
|
|
errors.append(f"第 {i} 项验证时出错: {e}")
|
|
|
|
result = {
|
|
"valid": len(errors) == 0,
|
|
"total": len(input_data),
|
|
"errors": errors,
|
|
"warnings": warnings
|
|
}
|
|
|
|
if self._context:
|
|
self._context.set_shared_data("validation_result", result)
|
|
self._context.publish_event("data_validated", result)
|
|
|
|
self.status = ModuleStatus.READY
|
|
return result
|
|
|
|
|
|
class StatisticsAnalyzerModule(BaseModule):
|
|
"""
|
|
统计分析模块
|
|
|
|
负责计算数据的统计指标。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
super().__init__(config)
|
|
self._metadata = ModuleMetadata(
|
|
name="statistics_analyzer",
|
|
version="1.0.0",
|
|
module_type=ModuleType.ANALYZER,
|
|
description="计算数据统计指标",
|
|
dependencies=["csv_data_loader"],
|
|
author="CC4SI"
|
|
)
|
|
|
|
def execute(self, input_data: Any) -> Any:
|
|
"""
|
|
执行统计分析
|
|
|
|
Args:
|
|
input_data: 待分析的数据
|
|
|
|
Returns:
|
|
统计结果
|
|
"""
|
|
self.status = ModuleStatus.RUNNING
|
|
|
|
if not isinstance(input_data, list) or len(input_data) == 0:
|
|
return {"error": "没有可分析的数据"}
|
|
|
|
# 提取数值字段
|
|
values = [item.get("value", 0) for item in input_data if "value" in item]
|
|
|
|
if not values:
|
|
return {"error": "没有找到可分析的数值"}
|
|
|
|
import statistics
|
|
|
|
result = {
|
|
"count": len(values),
|
|
"mean": statistics.mean(values),
|
|
"median": statistics.median(values),
|
|
"stdev": statistics.stdev(values) if len(values) > 1 else 0,
|
|
"min": min(values),
|
|
"max": max(values),
|
|
"sum": sum(values)
|
|
}
|
|
|
|
if self._context:
|
|
self._context.set_shared_data("statistics", result)
|
|
self._context.publish_event("analysis_complete", result)
|
|
|
|
self.status = ModuleStatus.READY
|
|
return result
|
|
|
|
|
|
class ReportExporterModule(BaseModule):
|
|
"""
|
|
报告导出模块
|
|
|
|
负责生成分析报告。
|
|
"""
|
|
|
|
def __init__(self, config: Dict[str, Any] = None):
|
|
super().__init__(config)
|
|
self._metadata = ModuleMetadata(
|
|
name="report_exporter",
|
|
version="1.0.0",
|
|
module_type=ModuleType.EXPORTER,
|
|
description="生成分析报告",
|
|
dependencies=["statistics_analyzer"],
|
|
author="CC4SI"
|
|
)
|
|
|
|
def execute(self, input_data: Any) -> Any:
|
|
"""
|
|
生成报告
|
|
|
|
Args:
|
|
input_data: 统计结果或其他数据
|
|
|
|
Returns:
|
|
报告字符串
|
|
"""
|
|
self.status = ModuleStatus.RUNNING
|
|
|
|
report_lines = [
|
|
"=" * 60,
|
|
"空间数据分析报告",
|
|
"=" * 60,
|
|
""
|
|
]
|
|
|
|
# 从上下文获取数据
|
|
if self._context:
|
|
validation = self._context.get_shared_data("validation_result")
|
|
statistics = self._context.get_shared_data("statistics")
|
|
|
|
if validation:
|
|
report_lines.extend([
|
|
"数据验证结果:",
|
|
f" 总数: {validation.get('total', 0)}",
|
|
f" 有效: {validation.get('valid', False)}",
|
|
f" 错误数: {len(validation.get('errors', []))}",
|
|
""
|
|
])
|
|
|
|
if statistics:
|
|
report_lines.extend([
|
|
"统计分析结果:",
|
|
f" 样本数: {statistics.get('count', 0)}",
|
|
f" 均值: {statistics.get('mean', 0):.2f}",
|
|
f" 中位数: {statistics.get('median', 0):.2f}",
|
|
f" 标准差: {statistics.get('stdev', 0):.2f}",
|
|
f" 最小值: {statistics.get('min', 0)}",
|
|
f" 最大值: {statistics.get('max', 0)}",
|
|
""
|
|
])
|
|
|
|
report_lines.append("=" * 60)
|
|
|
|
report = "\n".join(report_lines)
|
|
|
|
if self._context:
|
|
self._context.publish_event("report_generated", report)
|
|
|
|
self.status = ModuleStatus.READY
|
|
return report
|
|
|
|
|
|
# ============================================================================
|
|
# 模块系统管理器
|
|
# ============================================================================
|
|
|
|
class ModularSystem:
|
|
"""
|
|
模块化系统管理器
|
|
|
|
负责管理模块的生命周期和模块间通信。
|
|
"""
|
|
|
|
def __init__(self, name: str = "模块化空间智能系统"):
|
|
self.name = name
|
|
self.context = ModuleContext()
|
|
self._pipeline: List[str] = [] # 处理流程
|
|
|
|
def register_module(self, module: IModule, alias: str = None) -> bool:
|
|
"""
|
|
注册模块到系统
|
|
|
|
Args:
|
|
module: 模块实例
|
|
alias: 模块别名 (可选)
|
|
|
|
Returns:
|
|
是否注册成功
|
|
"""
|
|
name = alias or module.metadata.name
|
|
return self.context.register_module(name, module)
|
|
|
|
def initialize_all(self) -> bool:
|
|
"""初始化所有模块"""
|
|
print(f"\n初始化 {self.name}...")
|
|
|
|
success = True
|
|
for name, module in self.context._modules.items():
|
|
if not module.initialize(self.context):
|
|
print(f"模块 '{name}' 初始化失败")
|
|
success = False
|
|
|
|
return success
|
|
|
|
def define_pipeline(self, module_names: List[str]) -> None:
|
|
"""
|
|
定义处理流程
|
|
|
|
Args:
|
|
module_names: 按顺序执行的模块名称列表
|
|
"""
|
|
self._pipeline = module_names
|
|
print(f"定义处理流程: {' -> '.join(module_names)}")
|
|
|
|
def execute(self, input_data: Any = None) -> Any:
|
|
"""
|
|
执行处理流程
|
|
|
|
Args:
|
|
input_data: 输入数据
|
|
|
|
Returns:
|
|
最终输出结果
|
|
"""
|
|
if not self._pipeline:
|
|
print("错误: 没有定义处理流程")
|
|
return None
|
|
|
|
print(f"\n执行处理流程...")
|
|
current_data = input_data
|
|
|
|
for module_name in self._pipeline:
|
|
module = self.context.get_module(module_name)
|
|
if not module:
|
|
print(f"错误: 找不到模块 '{module_name}'")
|
|
return None
|
|
|
|
print(f" -> 执行模块: {module.metadata.name}")
|
|
current_data = module.execute(current_data)
|
|
|
|
# 如果模块返回错误，终止流程
|
|
if isinstance(current_data, dict) and current_data.get("error"):
|
|
print(f" 模块 '{module_name}' 返回错误: {current_data['error']}")
|
|
return current_data
|
|
|
|
return current_data
|
|
|
|
def shutdown_all(self) -> None:
|
|
"""关闭所有模块"""
|
|
print(f"\n关闭 {self.name}...")
|
|
for module in self.context._modules.values():
|
|
module.shutdown()
|
|
|
|
def print_system_info(self) -> None:
|
|
"""打印系统信息"""
|
|
print(f"\n{'='*60}")
|
|
print(f"系统: {self.name}")
|
|
print(f"{'='*60}")
|
|
print(f"已注册模块数: {len(self.context._modules)}")
|
|
|
|
for name, module in self.context._modules.items():
|
|
print(f" - {name:20s} [{module.metadata.module_type.value:15s}] {module.metadata.name}")
|
|
if module.metadata.dependencies:
|
|
print(f" 依赖: {', '.join(module.metadata.dependencies)}")
|
|
|
|
if self._pipeline:
|
|
print(f"\n处理流程: {' -> '.join(self._pipeline)}")
|
|
else:
|
|
print(f"\n处理流程: 未定义")
|
|
|
|
print(f"{'='*60}\n")
|
|
|
|
|
|
# ============================================================================
|
|
# 事件处理示例
|
|
# ============================================================================
|
|
|
|
def setup_event_handlers(system: ModularSystem):
|
|
"""设置事件处理器"""
|
|
|
|
def on_data_loaded(count):
|
|
print(f" [事件] 数据加载完成，共 {count} 条记录")
|
|
|
|
def on_data_validated(result):
|
|
status = "通过" if result.get("valid") else "失败"
|
|
print(f" [事件] 数据验证{status}，错误: {len(result.get('errors', []))}")
|
|
|
|
def on_analysis_complete(result):
|
|
print(f" [事件] 分析完成，均值: {result.get('mean', 0):.2f}")
|
|
|
|
system.context.subscribe_event("data_loaded", on_data_loaded)
|
|
system.context.subscribe_event("data_validated", on_data_validated)
|
|
system.context.subscribe_event("analysis_complete", on_analysis_complete)
|
|
|
|
|
|
# ============================================================================
|
|
# 主程序
|
|
# ============================================================================
|
|
|
|
def main():
|
|
"""主程序 - 演示模块化系统的使用"""
|
|
|
|
print("="*70)
|
|
print("模块化系统示例演示")
|
|
print("="*70)
|
|
|
|
# 1. 创建系统
|
|
print("\n[步骤 1] 创建模块化系统")
|
|
system = ModularSystem("空间数据分析系统")
|
|
|
|
# 2. 注册模块
|
|
print("\n[步骤 2] 注册模块")
|
|
system.register_module(CSVDataLoaderModule())
|
|
system.register_module(DataValidationModule())
|
|
system.register_module(StatisticsAnalyzerModule())
|
|
system.register_module(ReportExporterModule())
|
|
|
|
# 3. 设置事件处理
|
|
print("\n[步骤 3] 设置事件处理")
|
|
setup_event_handlers(system)
|
|
|
|
# 4. 初始化所有模块
|
|
print("\n[步骤 4] 初始化模块")
|
|
if not system.initialize_all():
|
|
print("初始化失败，退出")
|
|
return
|
|
|
|
# 5. 定义处理流程
|
|
print("\n[步骤 5] 定义处理流程")
|
|
system.define_pipeline([
|
|
"csv_data_loader",
|
|
"data_validator",
|
|
"statistics_analyzer",
|
|
"report_exporter"
|
|
])
|
|
|
|
# 6. 打印系统信息
|
|
print("\n[步骤 6] 系统信息")
|
|
system.print_system_info()
|
|
|
|
# 7. 执行处理流程
|
|
print("\n[步骤 7] 执行处理流程")
|
|
result = system.execute("sample_data.csv")
|
|
|
|
# 8. 输出结果
|
|
print("\n[步骤 8] 最终结果")
|
|
if isinstance(result, str):
|
|
print(result)
|
|
|
|
# 9. 清理
|
|
print("\n[步骤 9] 清理资源")
|
|
system.shutdown_all()
|
|
|
|
print("\n" + "="*70)
|
|
print("演示完成!")
|
|
print("="*70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|