forked from masonhuang/cluster4npu
abin
5aa374625f
- Add .autoflow/ with health check, PRD, Design Doc, TDD, progress tracking - Add tests/conftest.py with PyQt5/KP SDK stubs for unit testing - Add pytest config to pyproject.toml (pythonpath, import-mode, test naming) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1150 lines
38 KiB
Markdown
1150 lines
38 KiB
Markdown
# 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`
|
||
|
||
#### 公開介面
|
||
|
||
```python
|
||
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` 類別
|
||
|
||
```python
|
||
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` 回傳格式
|
||
|
||
```python
|
||
{
|
||
'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` 類別
|
||
|
||
```python
|
||
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` 函式
|
||
|
||
```python
|
||
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` 類別
|
||
|
||
```python
|
||
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`
|
||
|
||
#### 資料結構
|
||
|
||
```python
|
||
@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` 類別
|
||
|
||
```python
|
||
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 不同）
|
||
|
||
```python
|
||
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)` 判斷條件：
|
||
|
||
```python
|
||
# 有效結果（計入 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`
|
||
|
||
#### 結果資料類別
|
||
|
||
```python
|
||
@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）
|
||
|
||
```python
|
||
class PostProcessType(Enum):
|
||
FIRE_DETECTION = "fire_detection" # 二元分類（火焰偵測）
|
||
YOLO_V3 = "yolo_v3" # 物件偵測
|
||
YOLO_V5 = "yolo_v5" # 物件偵測（使用參考實作）
|
||
CLASSIFICATION = "classification" # 一般分類
|
||
RAW_OUTPUT = "raw_output" # 原始 numpy 輸出
|
||
```
|
||
|
||
#### `PostProcessorOptions` 設定
|
||
|
||
```python
|
||
@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` 處理流程
|
||
|
||
```python
|
||
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）
|
||
|
||
```python
|
||
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 多裝置效能測試，計算加速倍數。
|
||
|
||
```python
|
||
@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 歷史記錄，支援回歸測試追蹤。
|
||
|
||
```python
|
||
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）：**
|
||
```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。
|
||
|
||
```python
|
||
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 的對話框，顯示進度與結果。
|
||
|
||
```python
|
||
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 的裝置管理能力，提供視覺化分配介面所需的資料。
|
||
|
||
```python
|
||
@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 的即時狀態與分配情況。
|
||
|
||
```python
|
||
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 秒。
|
||
|
||
```python
|
||
@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 執行統計，產生可執行的優化建議。
|
||
|
||
```python
|
||
@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 範本。
|
||
|
||
```python
|
||
@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 資料結構
|
||
|
||
```python
|
||
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`
|
||
|
||
```python
|
||
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`）
|
||
|
||
```python
|
||
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`
|
||
|
||
```python
|
||
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` 測試重點
|
||
|
||
```python
|
||
# 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` 測試重點
|
||
|
||
```python
|
||
# 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` 後處理測試重點
|
||
|
||
```python
|
||
# 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 整合測試重點
|
||
|
||
```python
|
||
# 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：
|
||
|
||
```python
|
||
# 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
|
||
```
|