cluster4npu/.autoflow/04-architecture/TDD.md

TDD — Cluster4NPU UI

作者：Architect Agent

狀態：Draft

最後更新：2026-04-05

版本對應：v0.0.3（developer 分支）

---

1. 模組清單

模組路徑	類別/主要函式	狀態
core/pipeline.py	PipelineStage、analyze_pipeline_stages、validate_pipeline_structure、get_pipeline_summary	已完成
core/nodes/base_node.py	BaseNodeWithProperties、create_node_property_widget	已完成
core/nodes/input_node.py	InputNode	已完成
core/nodes/model_node.py	ModelNode	已完成
core/nodes/preprocess_node.py	PreprocessNode	已完成
core/nodes/postprocess_node.py	PostprocessNode	已完成
core/nodes/output_node.py	OutputNode	已完成
core/nodes/simple_input_node.py	SimpleInputNode	已完成
core/nodes/exact_nodes.py	ExactInputNode、ExactModelNode 等	已完成
core/functions/InferencePipeline.py	StageConfig、PipelineData、PipelineStage（執行）、InferencePipeline	已完成
core/functions/Multidongle.py	MultiDongle、PreProcessor、PostProcessor、PostProcessorOptions、結果資料類別	已完成
core/functions/camera_source.py	相機輸入來源	已完成
core/functions/video_source.py	影片輸入來源	已完成
core/functions/result_handler.py	推論結果處理	已完成
core/functions/mflow_converter.py	.mflow 格式轉換	已完成
core/functions/workflow_orchestrator.py	工作流程協調	已完成
core/functions/yolo_v5_postprocess_reference.py	YOLOv5 後處理（參考實作）	已完成
example_utils/	ByteTrack 物件追蹤等後處理工具	已完成
main.py	SingleInstance、setup_application、main	已完成
ui/windows/login.py	DashboardLogin	已完成
ui/windows/dashboard.py	DashboardWindow	已完成
ui/windows/pipeline_editor.py	PipelineEditor	已完成
ui/components/	NodePalette、PropertiesWidget、通用 Widget	已完成
ui/dialogs/	部署、效能、Stage 設定對話框	已完成
config/settings.py	應用程式設定	已完成
config/theme.py	Qt 主題套用	已完成
core/performance/benchmarker.py	PerformanceBenchmarker（待開發）	待開發
core/performance/history.py	PerformanceHistory（待開發）	待開發
core/device/device_manager.py	DeviceManager（待開發）	待開發
core/optimization/engine.py	OptimizationEngine（待開發）	待開發
ui/components/performance_dashboard.py	PerformanceDashboard（待開發）	待開發
ui/components/device_management_panel.py	DeviceManagementPanel（待開發）	待開發

---

2. 現有模組技術規格

2.1 core/pipeline.py

公開介面

def analyze_pipeline_stages(node_graph) -> List[PipelineStage]
def get_stage_count(node_graph) -> int
def validate_pipeline_structure(node_graph) -> Tuple[bool, str]
def get_pipeline_summary(node_graph) -> Dict[str, Any]
def find_connected_nodes(node, visited=None, direction='forward') -> List

PipelineStage 類別

class PipelineStage:
    stage_id: int
    model_node: ModelNode
    preprocess_nodes: List[PreprocessNode]
    postprocess_nodes: List[PostprocessNode]
    input_connections: list
    output_connections: list

    def add_preprocess_node(node: PreprocessNode) -> None
    def add_postprocess_node(node: PostprocessNode) -> None
    def get_stage_config() -> Dict[str, Any]
    # 回傳格式：
    # {
    #   'stage_id': int,
    #   'model_config': Dict,        # from ModelNode.get_inference_config()
    #   'preprocess_configs': List,  # from PreprocessNode.get_preprocessing_config()
    #   'postprocess_configs': List  # from PostprocessNode.get_postprocessing_config()
    # }
    def validate_stage() -> Tuple[bool, str]

get_pipeline_summary 回傳格式

{
    'stage_count': int,
    'valid': bool,
    'error': Optional[str],
    'stages': List[Dict],        # 各 Stage 的 get_stage_config() 結果
    'total_nodes': int,
    'input_nodes': int,
    'output_nodes': int,
    'model_nodes': int,
    'preprocess_nodes': int,
    'postprocess_nodes': int
}

依賴關係

• 輸入：NodeGraphQt NodeGraph 物件
• 依賴：core/nodes/*.py（型別識別）
• 輸出：純 Python 資料結構（Dict、List）

節點識別策略（多重 fallback）

下列任一條件成立即判定為 ModelNode：

1. node.__identifier__ 包含 "model"
2. node.type_ 包含 "model"
3. node.NODE_NAME 包含 "model"
4. type(node) 名稱包含 "model"
5. hasattr(node, 'get_inference_config') 為 True
6. type(node) 名稱包含 "exactmodel"

---

2.2 core/nodes/base_node.py

BaseNodeWithProperties 類別

class BaseNodeWithProperties(NodeGraphQt.BaseNode):
    # 內部狀態
    _property_options: Dict[str, Any]
    _property_validators: Dict[str, callable]
    _business_properties: Dict[str, Any]

    # 主要方法
    def create_business_property(
        name: str,
        default_value: Any,
        options: Optional[Dict[str, Any]] = None
    ) -> None

    def validate_property(name: str, value: Any) -> bool
    # 驗證規則（options 格式）：
    # - {'min': X, 'max': Y}  → 數值範圍驗證
    # - [choice1, choice2]    → 選項列表驗證
    # - {'type': 'file_path'} → 檔案路徑（不驗證存在）

    def get_node_config() -> Dict[str, Any]
    # 回傳格式：
    # {
    #   'type': str,         # 類別名稱
    #   'name': str,         # 節點顯示名稱
    #   'properties': Dict,  # 所有 business properties
    #   'position': Tuple    # (x, y)
    # }

    def load_node_config(config: Dict[str, Any]) -> None
    def update_business_property(name: str, value: Any) -> bool  # 含驗證

create_node_property_widget 函式

def create_node_property_widget(
    node: BaseNodeWithProperties,
    prop_name: str,
    prop_value: Any,
    options: Optional[Dict[str, Any]] = None
) -> QWidget

prop_value 型別	options 條件	產生 Widget
Any	options.get('type') == 'file_path'	QPushButton（開啟檔案對話框）
bool	—	QCheckBox
Any	isinstance(options, list)	QComboBox
int	—	QSpinBox（min/max 取自 options）
float	—	QDoubleSpinBox（min/max/decimals/step）
str（預設）	—	QLineEdit

---

2.3 core/nodes/model_node.py

ModelNode 類別

class ModelNode(BaseNodeWithProperties):
    __identifier__ = 'com.cluster.model_node'
    NODE_NAME = 'Model Node'

    # 連接埠
    # input: 'input'（single，橙色 #FF8C00）
    # output: 'output'（綠色 #00FF00）
    # 節點顏色：RGB(65, 84, 102)

    def validate_configuration() -> Tuple[bool, str]
    # 驗證規則：
    # - model_path 不能為空
    # - dongle_series 必須為 '520', '720', '1080', 'Custom' 之一
    # - num_dongles 必須為 int 且 >= 1

    def get_inference_config() -> Dict[str, Any]
    # 回傳格式：
    # {
    #   'node_id': str,
    #   'node_name': str,
    #   'model_path': str,
    #   'dongle_series': str,
    #   'num_dongles': int,
    #   'port_id': str,
    #   'batch_size': int,
    #   'max_queue_size': int,
    #   'enable_preprocessing': bool,
    #   'enable_postprocessing': bool
    # }

    def get_hardware_requirements() -> Dict[str, Any]
    # 包含 dongle_series, num_dongles, port_id,
    # estimated_memory (MB), estimated_power (W)

ModelNode 屬性規格

屬性名稱	型別	預設值	驗證
model_path	file_path	''	非空（執行前）
dongle_series	choice	'520'	必須為 520/720/1080/Custom
num_dongles	int	1	1–16
port_id	str	''	無（auto 接受）
batch_size	int	1	1–32
max_queue_size	int	10	1–100
enable_preprocessing	bool	True	—
enable_postprocessing	bool	True	—

---

2.4 core/functions/InferencePipeline.py

資料結構

@dataclass
class StageConfig:
    stage_id: str
    port_ids: List[int]
    scpu_fw_path: str
    ncpu_fw_path: str
    model_path: str
    upload_fw: bool
    max_queue_size: int = 50
    multi_series_config: Optional[Dict[str, Any]] = None
    input_preprocessor: Optional[PreProcessor] = None
    output_postprocessor: Optional[PostProcessor] = None
    stage_preprocessor: Optional[PreProcessor] = None
    stage_postprocessor: Optional[PostProcessor] = None

@dataclass
class PipelineData:
    data: Any
    metadata: Dict[str, Any]  # 包含 start_timestamp, end_timestamp, total_processing_time
    stage_results: Dict[str, Any]  # key = stage_id
    pipeline_id: str               # 格式："pipeline_{counter}"
    timestamp: float

InferencePipeline 類別

class InferencePipeline:
    def __init__(
        stage_configs: List[StageConfig],
        final_postprocessor: Optional[PostProcessor] = None,
        pipeline_name: str = "InferencePipeline"
    )

    # 生命週期
    def initialize() -> None    # 初始化所有 Stage（Sequential）
    def start() -> None         # 啟動 Coordinator Thread + 所有 Stage Workers
    def stop() -> None          # 優雅停止（Sentinel 模式 + join）

    # 資料 I/O
    def put_data(data: Any, timeout: float = 1.0) -> bool
    # 若輸入佇列已滿：捨棄最舊的幀（即時性優先）

    def get_result(timeout: float = 0.1) -> Optional[PipelineData]

    # 回調設定
    def set_result_callback(callback: Callable[[PipelineData], None]) -> None
    def set_error_callback(callback: Callable[[PipelineData], None]) -> None
    def set_stats_callback(callback: Callable[[Dict[str, Any]], None]) -> None

    # 效能
    def get_current_fps() -> float
    # 計算公式：completed_counter / (now - fps_start_time)
    # fps_start_time 設定時機：第一個有效結果完成時

    def get_pipeline_statistics() -> Dict[str, Any]
    # 回傳格式：
    # {
    #   'pipeline_name': str,
    #   'total_stages': int,
    #   'pipeline_input_submitted': int,
    #   'pipeline_completed': int,
    #   'pipeline_errors': int,
    #   'pipeline_input_queue_size': int,
    #   'pipeline_output_queue_size': int,
    #   'current_fps': float,
    #   'stage_statistics': List[Dict]  # 每個 Stage 的統計
    # }

    def start_stats_reporting(interval: float = 5.0) -> None

PipelineStage（執行層，與 pipeline.py 中的 PipelineStage 不同）

class PipelineStage:  # 在 InferencePipeline.py 中
    def put_data(data: PipelineData, timeout: float = 1.0) -> bool
    def get_result(timeout: float = 0.1) -> Optional[PipelineData]
    def get_statistics() -> Dict[str, Any]
    # 回傳格式：
    # {
    #   'stage_id': str,
    #   'processed_count': int,
    #   'error_count': int,
    #   'avg_processing_time': float,
    #   'input_queue_size': int,
    #   'output_queue_size': int,
    #   'multidongle_stats': Dict  # 來自 MultiDongle.get_statistics()
    # }

佇列規格

佇列	maxsize	滿時策略
pipeline_input_queue	100	捨棄最舊幀（即時性）
Stage input_queue	StageConfig.max_queue_size（預設 50）	捨棄並 drop
Stage output_queue	StageConfig.max_queue_size（預設 50）	捨棄最舊並替換
pipeline_output_queue	50	捨棄最舊結果（預防性清理）

有效推論結果判斷邏輯

_has_valid_inference_result(pipeline_data) 判斷條件：

# 有效結果（計入 FPS / 放入輸出佇列）：
# - Tuple (prob, result_str)：prob is not None and result_str not in ['Processing']
# - Dict：status 不為 "processing"/"async"，且 result 不為 "Processing"

# 無效結果（不計入 FPS，丟棄）：
# - {'status': 'processing'} 或 {'status': 'async'}
# - {'result': 'Processing'}
# - 空結果

---

2.5 core/functions/Multidongle.py

結果資料類別

@dataclass
class BoundingBox:
    x1: int; y1: int; x2: int; y2: int
    score: float
    class_num: int
    class_name: str

@dataclass
class ObjectDetectionResult:
    class_count: int
    box_count: int
    box_list: List[BoundingBox]
    # Letterbox 映射資訊（用於座標還原）：
    model_input_width: int; model_input_height: int
    resized_img_width: int; resized_img_height: int
    pad_left: int; pad_top: int; pad_right: int; pad_bottom: int

@dataclass
class ClassificationResult:
    probability: float
    class_name: str
    class_num: int
    confidence_threshold: float
    # property: is_positive -> probability > confidence_threshold

後處理類型（PostProcessType Enum）

class PostProcessType(Enum):
    FIRE_DETECTION = "fire_detection"   # 二元分類（火焰偵測）
    YOLO_V3 = "yolo_v3"                 # 物件偵測
    YOLO_V5 = "yolo_v5"                 # 物件偵測（使用參考實作）
    CLASSIFICATION = "classification"   # 一般分類
    RAW_OUTPUT = "raw_output"           # 原始 numpy 輸出

PostProcessorOptions 設定

@dataclass
class PostProcessorOptions:
    postprocess_type: PostProcessType = PostProcessType.FIRE_DETECTION
    threshold: float = 0.5             # 信心度閾值
    class_names: List[str] = []        # 類別名稱列表
    nms_threshold: float = 0.45        # NMS 閾值（YOLO）
    max_detections_per_class: int = 100

PreProcessor 處理流程

class PreProcessor(DataProcessor):
    def process(frame: np.ndarray, target_size: tuple, target_format: str) -> np.ndarray:
        # Step 1: resize（預設使用 cv2.resize）
        # Step 2: format convert
        #   - 'BGR565'   → cv2.cvtColor(frame, COLOR_BGR2BGR565)
        #   - 'RGB8888'  → cv2.cvtColor(frame, COLOR_BGR2RGBA)
        #   - 其他格式   → 直接回傳

裝置算力規格（DongleSeriesSpec）

SERIES_SPECS = {
    "KL520": {"product_id": 0x100,  "gops": 2},
    "KL720": {"product_id": 0x720,  "gops": 28},
    "KL630": {"product_id": 0x630,  "gops": 400},
    "KL730": {"product_id": 0x730,  "gops": 1600},
}

---

3. 待開發功能技術規格

Phase 1：效能視覺化 / Benchmarking

3.1.1 PerformanceBenchmarker（core/performance/benchmarker.py）

職責： 自動執行單裝置 vs 多裝置效能測試，計算加速倍數。

@dataclass
class BenchmarkConfig:
    pipeline_config: List[StageConfig]   # 來自 UI 的 Pipeline 設定
    test_duration_seconds: float = 30.0  # 測試持續時間
    warmup_frames: int = 50              # 熱機幀數（不計入統計）
    test_input_source: str               # 測試輸入（影片/相機）

@dataclass
class BenchmarkResult:
    mode: str                            # 'sequential' | 'parallel'
    fps: float
    avg_latency_ms: float
    p95_latency_ms: float
    total_frames: int
    timestamp: float
    device_config: Dict[str, Any]        # 裝置分配設定

class PerformanceBenchmarker:
    def run_sequential_benchmark(config: BenchmarkConfig) -> BenchmarkResult
    # 強制使用單一 Dongle 執行 Pipeline

    def run_parallel_benchmark(config: BenchmarkConfig) -> BenchmarkResult
    # 使用全部可用 Dongle 執行 Pipeline

    def calculate_speedup(
        seq: BenchmarkResult,
        par: BenchmarkResult
    ) -> float
    # 計算公式：par.fps / seq.fps

    def run_full_benchmark(config: BenchmarkConfig) -> Tuple[BenchmarkResult, BenchmarkResult, float]
    # 回傳：(sequential_result, parallel_result, speedup)

執行序列：

1. 暖機（warmup_frames 幀，不計入）
2. 循序模式：強制單一 Dongle → 收集 test_duration_seconds 秒資料
3. 清空佇列 + 重啟 Pipeline
4. 平行模式：使用全部 Dongle → 收集相同時間資料
5. 計算加速倍數

3.1.2 PerformanceHistory（core/performance/history.py）

職責： 本地儲存 Benchmark 歷史記錄，支援回歸測試追蹤。

class PerformanceHistory:
    def __init__(storage_path: str = "~/.cluster4npu/benchmark_history.json")

    def record(result: BenchmarkResult) -> None
    def get_history(
        limit: int = 50,
        mode: Optional[str] = None
    ) -> List[BenchmarkResult]

    def get_regression_report(
        baseline_id: str,
        compare_id: str
    ) -> Dict[str, Any]
    # 比較兩次測試的 FPS/延遲差異

儲存格式（JSON）：

{
  "records": [
    {
      "id": "benchmark_20260405_143022",
      "mode": "parallel",
      "fps": 45.2,
      "avg_latency_ms": 22.1,
      "p95_latency_ms": 35.0,
      "total_frames": 1356,
      "timestamp": 1743856222.0,
      "device_config": {"KL720": 2}
    }
  ]
}

3.1.3 PerformanceDashboard（ui/components/performance_dashboard.py）

職責： 顯示即時 FPS 和延遲折線圖，更新頻率 >= 1 Hz。

class PerformanceDashboard(QWidget):
    # Qt Signals
    update_requested = pyqtSignal(dict)  # 接收來自 InferencePipeline 的統計資料

    def __init__(parent: Optional[QWidget] = None)

    def update_stats(stats: Dict[str, Any]) -> None
    # stats 格式與 InferencePipeline.get_pipeline_statistics() 相同

    def reset() -> None  # 清空圖表歷史

    def set_display_window(seconds: int = 60) -> None
    # 設定圖表顯示的時間視窗（秒）

效能約束：

• 圖表更新不得使推論 FPS 下降超過 5%（使用 QTimer 限速 + 背景執行緒計算）
• 建議使用 pyqtgraph（效能優於 matplotlib）

3.1.4 Benchmark 觸發 UI（ui/dialogs/benchmark_dialog.py）

職責： 一鍵啟動 Benchmark 的對話框，顯示進度與結果。

class BenchmarkDialog(QDialog):
    def __init__(
        parent: QWidget,
        pipeline_config: List[StageConfig]
    )

    # 顯示：
    # - 進度條（熱機 / 循序測試 / 平行測試）
    # - 即時 FPS 顯示
    # - 完成後：加速倍數（大字體，如 "3.2x FASTER"）
    # - 循序 vs 平行的 FPS 與延遲對比表

---

Phase 2：裝置管理

3.2.1 DeviceManager（core/device/device_manager.py）

職責： 擴展 MultiDongle 的裝置管理能力，提供視覺化分配介面所需的資料。

@dataclass
class DeviceInfo:
    device_id: str              # 唯一識別碼（如 USB port 位置）
    series: str                 # "KL520" | "KL720" | "KL1080"
    product_id: int             # 來自 DongleSeriesSpec
    status: str                 # "online" | "offline" | "busy"
    gops: int                   # 算力（來自 DongleSeriesSpec）
    assigned_stage: Optional[str]  # 目前分配的 Stage ID
    current_fps: float          # 當前推論吞吐量
    utilization_pct: float      # 使用率百分比（0.0–100.0）

@dataclass
class DeviceHealth:
    device_id: str
    temperature_celsius: Optional[float]  # 如 SDK 支援
    error_count: int
    last_error: Optional[str]
    uptime_seconds: float

class DeviceManager:
    def scan_devices() -> List[DeviceInfo]
    # 呼叫 Kneron KP SDK 掃描已連接的 Dongle

    def get_device_health(device_id: str) -> DeviceHealth

    def assign_device(device_id: str, stage_id: str) -> bool
    # 若 device_id 離線或已分配至其他 Stage → 回傳 False

    def unassign_device(device_id: str) -> bool

    def get_load_balance_recommendation(
        stages: List[str]
    ) -> Dict[str, str]
    # 依據 DongleSeriesSpec.gops 分配裝置給各 Stage
    # 回傳格式：{stage_id: device_id}

    def get_device_statistics() -> Dict[str, DeviceInfo]
    # 回傳所有裝置的即時狀態

負載平衡演算法（初版）：

• 計算每個 Stage 的推論需求（GOPS）
• 依 Dongle 的 GOPS 算力做比例分配
• 優先分配高算力 Dongle 給第一個（或最繁忙的）Stage

3.2.2 DeviceManagementPanel（ui/components/device_management_panel.py）

職責： 顯示所有 NPU Dongle 的即時狀態與分配情況。

class DeviceManagementPanel(QWidget):
    device_assignment_changed = pyqtSignal(str, str)  # (device_id, stage_id)

    def __init__(device_manager: DeviceManager, parent: Optional[QWidget] = None)

    def refresh() -> None
    # 重新掃描裝置並更新 UI

    def set_auto_refresh(interval_ms: int = 2000) -> None
    # 設定自動刷新間隔

UI 顯示內容：

• 每個 Dongle 一張卡片：型號、狀態指示燈、目前分配 Stage、即時 FPS
• 下拉選單允許手動更改分配
• 「自動平衡」按鈕呼叫 get_load_balance_recommendation()

3.2.3 瓶頸偵測（整合至 InferencePipeline）

觸發條件： 某 Stage 的 input_queue.qsize() > max_queue_size * 0.8 持續超過 5 秒。

@dataclass
class BottleneckAlert:
    stage_id: str
    queue_fill_rate: float       # 佇列使用率（0.0–1.0）
    suggested_action: str        # 如 "增加此 Stage 的 Dongle 數量"
    severity: str                # "warning" | "critical"

# 在 InferencePipeline 中新增：
def get_bottleneck_alerts() -> List[BottleneckAlert]
def set_bottleneck_callback(callback: Callable[[BottleneckAlert], None]) -> None

---

Phase 3：進階功能

3.3.1 OptimizationEngine（core/optimization/engine.py）

職責： 分析 Pipeline 執行統計，產生可執行的優化建議。

@dataclass
class OptimizationSuggestion:
    suggestion_id: str
    type: str               # "rebalance_devices" | "remove_node" | "add_devices" | "adjust_queue"
    description: str        # 使用者可讀的說明（避免技術術語）
    estimated_improvement_pct: float   # 預估改善百分比
    confidence: str         # "high" | "medium" | "low"
    action_params: Dict[str, Any]      # 執行建議所需的參數

class OptimizationEngine:
    def analyze_pipeline(
        stats: Dict[str, Any]   # 來自 InferencePipeline.get_pipeline_statistics()
    ) -> List[OptimizationSuggestion]

    def apply_suggestion(
        suggestion: OptimizationSuggestion,
        device_manager: DeviceManager
    ) -> bool

    def predict_performance(
        config: List[StageConfig],
        available_devices: List[DeviceInfo]
    ) -> Dict[str, float]
    # 回傳格式：
    # {
    #   'estimated_fps': float,
    #   'estimated_latency_ms': float,
    #   'confidence_range': Tuple[float, float]  # [min, max] FPS
    # }

優化規則（初版）：

1. rebalance_devices：若某 Stage 的佇列使用率 > 70%，建議將算力較高的 Dongle 分配給該 Stage
2. adjust_queue：若 avg_processing_time 差異超過 2 倍，建議調整佇列大小
3. add_devices：若所有 Dongle 使用率 > 85%，建議增加更多 Dongle

3.3.2 Pipeline 設定範本（core/templates/）

職責： 提供常見使用情境的預設 Pipeline 範本。

@dataclass
class PipelineTemplate:
    template_id: str
    name: str               # 如 "YOLOv5 物件偵測"
    description: str
    nodes: List[Dict]       # 節點定義（與 .mflow 格式相同）
    connections: List[Dict]

class TemplateManager:
    def get_builtin_templates() -> List[PipelineTemplate]
    # 至少提供 3 種範本：
    # 1. YOLOv5 物件偵測（Input → Preprocess → Model → Postprocess → Output）
    # 2. 火焰偵測分類（Input → Model → Postprocess → Output）
    # 3. 雙模型串接（Input → Model1 → Postprocess1 → Model2 → Postprocess2 → Output）

    def save_as_template(
        pipeline_config: Dict,
        name: str,
        description: str
    ) -> PipelineTemplate

    def load_template(template_id: str) -> PipelineTemplate

---

Phase 4：效能報告匯出（PDF/CSV）

3.4.1 模組位置

core/
└── performance/
    └── report_exporter.py   # ReportExporter 主類別、ReportData dataclass

UI 觸發入口：ui/dialogs/export_report_dialog.py（ExportReportDialog，讓使用者選擇格式與儲存路徑）

3.4.2 函式庫選型

用途	選用函式庫	理由
PDF 匯出	reportlab	功能完整、純 Python、無系統依賴、在 PyQt5 環境下相容性佳；支援表格、圖表嵌入、自訂版面
CSV 匯出	csv（標準庫）	零依賴、足夠滿足表格資料匯出需求
圖表截圖（嵌入 PDF）	pyqtgraph + QPixmap.grabWidget()	直接從現有 PerformanceDashboard 擷取折線圖，不需要重新繪製

不選 fpdf2 的原因： 嵌入 pyqtgraph 圖表截圖時需要額外影像處理步驟，reportlab 的 ImageReader 對 PIL/QPixmap 轉換更直接。

安裝指令（加入 requirements.txt）：

reportlab>=4.0.0

3.4.3 資料結構

from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import time

@dataclass
class DeviceSummary:
    """單一裝置的摘要資訊，來自 DeviceManager"""
    device_id: str
    product_name: str          # 如 "KL720"
    firmware_version: str
    is_active: bool

@dataclass
class ReportData:
    """
    報告所需的完整資料，由呼叫方（UI 層）從各模組收集後傳入 ReportExporter。
    設計為純資料容器，與 UI / SDK 解耦，方便單元測試。
    """
    # 報告基本資訊
    report_title: str = "效能測試報告"
    generated_at: float = field(default_factory=time.time)   # UNIX timestamp
    pipeline_name: str = ""        # 來自 .mflow 檔名或使用者命名

    # Benchmark 結果（來自 PerformanceBenchmarker.run_full_benchmark()）
    sequential_result: Optional[Any] = None   # BenchmarkResult
    parallel_result: Optional[Any] = None     # BenchmarkResult
    speedup: Optional[float] = None           # par.fps / seq.fps

    # 歷史記錄（來自 PerformanceHistory.get_history()）
    history_records: List[Any] = field(default_factory=list)  # List[BenchmarkResult]

    # 裝置資訊（來自 DeviceManager.get_all_devices()）
    devices: List[DeviceSummary] = field(default_factory=list)

    # 圖表截圖（由 UI 層在匯出前擷取）
    chart_image_bytes: Optional[bytes] = None  # PNG bytes，來自 PerformanceDashboard

3.4.4 核心類別設計：ReportExporter

from pathlib import Path
from typing import Optional
import csv, io, time

class ReportExporter:
    """
    負責將 ReportData 序列化為 PDF 或 CSV 檔案。
    無狀態設計（stateless）：每次匯出建立新實例或直接呼叫靜態方法。
    """

    # --- PDF 匯出 ---

    def export_pdf(
        self,
        data: ReportData,
        output_path: str | Path
    ) -> Path:
        """
        將完整效能報告匯出為 PDF。
        回傳實際寫入的檔案路徑。
        若 output_path 的父目錄不存在，自動建立。
        """

    def _build_cover_page(
        self,
        canvas,         # reportlab.pdfgen.canvas.Canvas
        data: ReportData
    ) -> None:
        """繪製封面：報告標題、生成時間、Pipeline 名稱、裝置清單"""

    def _build_benchmark_table(
        self,
        story: list,    # reportlab Flowable 清單
        data: ReportData
    ) -> None:
        """
        建立 Benchmark 結果對比表（reportlab Table）。
        欄位：指標 / 循序模式 / 平行模式 / 差異%
        指標：FPS、平均延遲(ms)、P95 延遲(ms)、總幀數
        """

    def _build_trend_chart(
        self,
        story: list,
        data: ReportData
    ) -> None:
        """
        若 data.chart_image_bytes 不為 None，將圖表 PNG 嵌入 PDF。
        若為 None，插入「無圖表資料」的提示文字。
        """

    def _build_history_table(
        self,
        story: list,
        data: ReportData
    ) -> None:
        """
        建立歷史記錄表（最多顯示 20 筆，超過則截斷並標注）。
        欄位：測試時間 / 模式 / FPS / 平均延遲(ms) / P95 延遲(ms)
        """

    def _build_device_info(
        self,
        story: list,
        data: ReportData
    ) -> None:
        """列出測試時連接的裝置清單：裝置 ID、型號、韌體版本、是否啟用"""

    # --- CSV 匯出 ---

    def export_csv(
        self,
        data: ReportData,
        output_path: str | Path
    ) -> Path:
        """
        將 Benchmark 結果與歷史記錄匯出為 CSV。
        CSV 包含兩個邏輯區塊（以空行分隔）：
        1. Benchmark 摘要（循序 vs 平行對比）
        2. 歷史記錄（每筆 BenchmarkResult 一行）
        回傳實際寫入的檔案路徑。
        """

    # --- 工廠方法（方便測試 mock） ---

    @staticmethod
    def _get_timestamp_str(ts: float) -> str:
        """將 UNIX timestamp 格式化為 'YYYY-MM-DD HH:MM:SS'（本地時間）"""

3.4.5 PDF 報告內容結構

區塊	內容	實作方法
封面	報告標題、生成時間（本地時間）、Pipeline 名稱、裝置數量摘要	_build_cover_page
Benchmark 結果表	循序 vs 平行的 FPS / 延遲對比，加速倍數以大字體標示（如「加速 3.2x」）	_build_benchmark_table
效能趨勢圖	從 PerformanceDashboard 擷取的 pyqtgraph 折線圖截圖（PNG 嵌入）	_build_trend_chart
歷史記錄表	最近 20 筆 Benchmark 記錄（時間、模式、FPS、延遲）	_build_history_table
裝置資訊	測試時連接的裝置清單（型號、韌體、是否啟用）	_build_device_info

PDF 頁面規格：A4（210×297mm），reportlab 預設單位 point（72pt = 1 inch）。

3.4.6 CSV 匯出格式

區塊 1：Benchmark 摘要

section,metric,sequential,parallel,diff_pct
benchmark_summary,fps,14.2,45.6,+221.1%
benchmark_summary,avg_latency_ms,70.4,21.9,-68.9%
benchmark_summary,p95_latency_ms,95.0,33.2,-65.1%
benchmark_summary,total_frames,426,1368,—
benchmark_summary,speedup,—,3.21x,—

區塊 2：歷史記錄（空一行後接續）

id,timestamp,mode,fps,avg_latency_ms,p95_latency_ms,total_frames
benchmark_20260405_143022,2026-04-05 14:30:22,parallel,45.2,22.1,35.0,1356
...

3.4.7 與現有模組的整合點

資料來源	取用方式	對應 ReportData 欄位
PerformanceBenchmarker.run_full_benchmark()	回傳 (sequential_result, parallel_result, speedup)	sequential_result, parallel_result, speedup
PerformanceHistory.get_history(limit=20)	回傳 List[BenchmarkResult]	history_records
DeviceManager.get_all_devices()	回傳裝置列表（型號、韌體等）	devices（轉為 DeviceSummary）
PerformanceDashboard（UI 層）	QPixmap 截圖後轉 PNG bytes	chart_image_bytes

整合原則： ReportExporter 本身不直接依賴上述任何模組。UI 層（ExportReportDialog）負責收集資料、組裝 ReportData，再傳給 ReportExporter。這樣 ReportExporter 可在無 PyQt5 環境（如 CI）下做單元測試。

3.4.8 UI 觸發入口（ui/dialogs/export_report_dialog.py）

class ExportReportDialog(QDialog):
    def __init__(
        self,
        parent: QWidget,
        benchmarker: PerformanceBenchmarker,
        history: PerformanceHistory,
        device_manager: DeviceManager,
        dashboard: PerformanceDashboard
    )

    def _collect_report_data(self) -> ReportData:
        """從各模組收集資料，組裝 ReportData"""

    def _on_export_pdf(self) -> None:
        """使用 QFileDialog 取得儲存路徑，呼叫 ReportExporter.export_pdf()"""

    def _on_export_csv(self) -> None:
        """使用 QFileDialog 取得儲存路徑，呼叫 ReportExporter.export_csv()"""

對話框 UI 元素：

• 匯出格式選擇（PDF / CSV / 兩者皆匯出）
• 儲存路徑輸入框（含「瀏覽」按鈕，QFileDialog）
• 進度指示（QProgressBar，PDF 生成時顯示）
• 匯出結果訊息（成功：顯示路徑；失敗：顯示錯誤訊息）

3.4.9 Phase 4 新增測試

測試檔案：tests/unit/test_report_exporter.py

def test_export_csv_creates_file_at_given_path():
    """export_csv() 應在指定路徑建立 CSV 檔案"""

def test_export_csv_contains_benchmark_summary_section():
    """CSV 應包含 benchmark_summary 區塊的 fps/latency 欄位"""

def test_export_csv_contains_history_section():
    """CSV 應包含歷史記錄區塊，行數等於 history_records 筆數"""

def test_export_csv_empty_history_produces_only_summary():
    """history_records 為空時，CSV 只輸出 Benchmark 摘要區塊"""

def test_export_csv_no_benchmark_result_raises_value_error():
    """sequential_result 或 parallel_result 為 None 時，應拋出 ValueError"""

def test_export_pdf_creates_file_at_given_path():
    """export_pdf() 應在指定路徑建立 PDF 檔案（不驗證內容，只驗證存在）"""

def test_export_pdf_auto_creates_parent_directory():
    """若輸出路徑的父目錄不存在，export_pdf() 應自動建立"""

def test_export_pdf_without_chart_image_does_not_raise():
    """chart_image_bytes 為 None 時，PDF 匯出不應拋出例外"""

def test_get_timestamp_str_format():
    """_get_timestamp_str 應回傳 'YYYY-MM-DD HH:MM:SS' 格式的字串"""

def test_report_data_defaults_are_sane():
    """ReportData 預設值：report_title 非空、generated_at 接近當下時間"""

---

4. 測試策略

4.1 測試架構

目標： 建立 pytest 框架，核心模組達 80% 以上覆蓋率。

測試目錄結構（建議）：

tests/
├── unit/
│   ├── test_pipeline_analysis.py      # core/pipeline.py 的 Stage 分析邏輯
│   ├── test_node_properties.py        # BaseNodeWithProperties 的屬性管理
│   ├── test_model_node.py             # ModelNode 的驗證與設定
│   ├── test_inference_pipeline.py     # InferencePipeline 的佇列、FPS 計算
│   ├── test_postprocessor.py          # PostProcessor 各類型的後處理邏輯
│   └── test_benchmarker.py            # PerformanceBenchmarker（Phase 1）
├── integration/
│   ├── test_pipeline_execution.py     # 完整 Pipeline 執行流程（Mock Hardware）
│   └── test_mflow_persistence.py      # .mflow 儲存與載入
└── e2e/
    └── test_ui_workflow.py            # UI 操作流程（需要顯示環境）

4.2 單元測試重點

core/pipeline.py 測試重點

# test_pipeline_analysis.py

def test_analyze_returns_empty_for_graph_without_input_output():
    """無 Input/Output 節點的 Graph 應回傳空 Stage 列表"""

def test_analyze_counts_model_nodes_as_stages():
    """Stage 數量應等於 ModelNode 數量"""

def test_stage_ordering_by_distance_from_input():
    """Stage 應依距離 Input 節點的遠近排序"""

def test_validate_pipeline_structure_requires_input_output_model():
    """缺少 Input、Output、或 Model 節點時，驗證應失敗"""

def test_node_detection_by_identifier():
    """is_model_node 應能透過 __identifier__ 識別 ModelNode"""

def test_node_detection_by_get_inference_config():
    """is_model_node 應能透過 get_inference_config 方法識別 ModelNode"""

core/functions/InferencePipeline.py 測試重點

# test_inference_pipeline.py

def test_fps_returns_zero_before_first_result():
    """初始化後 FPS 應為 0.0"""

def test_fps_excludes_processing_status_results():
    """status='processing' 的結果不應計入 FPS"""

def test_input_queue_drops_oldest_when_full():
    """輸入佇列滿時，最舊的幀應被捨棄"""

def test_pipeline_stop_gracefully():
    """stop() 應在 timeout 內完成（< 10 秒）"""

def test_result_callback_called_for_valid_results():
    """只有有效推論結果應觸發 result_callback"""

core/functions/Multidongle.py 後處理測試重點

# test_postprocessor.py

def test_fire_detection_above_threshold_returns_fire():
    """probability > threshold 應回傳 class_name='Fire'"""

def test_fire_detection_below_threshold_returns_no_fire():
    """probability <= threshold 應回傳 class_name='No Fire'"""

def test_classification_multiclass_returns_highest_confidence():
    """多類別分類應回傳最高信心度的類別"""

def test_yolo_v5_returns_object_detection_result():
    """YOLOv5 後處理應回傳 ObjectDetectionResult 型別"""

def test_yolo_v5_empty_output_returns_zero_boxes():
    """空推論輸出應回傳 box_count=0"""

4.3 整合測試重點

# test_pipeline_execution.py（使用 Mock Kneron SDK）

def test_pipeline_processes_frames_end_to_end():
    """影像幀應能完整通過所有 Stage"""

def test_pipeline_restarts_cleanly():
    """stop() 後重新 initialize() 和 start() 不應有狀態殘留"""

def test_multistage_pipeline_preserves_stage_order():
    """Stage 0 的結果應先於 Stage 1 的結果產出"""

4.4 效能測試目標

測試項目	目標
UI 互動回應	< 200ms（節點拖拽、屬性切換）
Pipeline 即時驗證延遲	< 100ms
PerformanceDashboard 更新 CPU 開銷	< 5% 推論 FPS 影響
應用程式啟動時間（含裝置偵測）	< 10 秒
Benchmark 一鍵到結果呈現	< 30 秒

4.5 Mock 策略

由於核心依賴 Kneron KP SDK（kp 模組）需要實際硬體，單元測試和整合測試應使用 Mock：

# conftest.py
import pytest
from unittest.mock import MagicMock, patch

@pytest.fixture
def mock_kp():
    """Mock Kneron KP SDK"""
    with patch('core.functions.Multidongle.kp') as mock:
        mock.scan_devices.return_value = [MagicMock(product_id=0x720)]
        mock.load_model.return_value = MagicMock()
        yield mock

@pytest.fixture
def mock_multidongle(mock_kp):
    """Mock MultiDongle with configurable inference results"""
    from unittest.mock import MagicMock
    dongle = MagicMock()
    dongle.get_latest_inference_result.return_value = (0.85, "Fire")
    dongle.model_input_shape = (224, 224)
    return dongle