jim800121chen
5a9c1b0140
轉檔時若 op 在目標 platform 不支援（如 Softmax@520），原本會白跑完 onnx+bie 兩階段、最後在 nef 的 batch_compile 拋難懂的 C++ backtrace（undefined CPU op）。 本次在 onnx 階段（拿到 out.onnx 後、進 bie 前）用 Kneron IP Evaluator （ktc ModelConfig.evaluate / ip_evaluator mode 0 會跑 compiler frontend） 早期偵測不支援 op，快速失敗並回清楚訊息。 - 新增 services/backends/precheck.py：run_precheck 掃 evaluate 的不支援訊號 （HardwareNotSupport / UnimplementedFeature / undefined CPU op / EmptyNode） 抽出 op 名、raise UnsupportedOperatorError；錯誤走既有 _push_done(fail) 回報 - services/workers/onnx/core.py：enable_precheck default-on；pre-check 與既有 evaluate 共用同一次 evaluate() 呼叫（避免跑兩次 compiler frontend） - error 訊息：「platform 520 不支援 operator Softmax，請改用支援的 platform （如 720/730）或將該 operator 移至 host 端後處理。」 - fail-open：只有明確訊號才擋、evaluate 環境問題無訊號則放行（不誤擋能跑的模型） - 17 tests pass（含多 op / dedup） follow-up（部署後、需 docker toolchain 環境）：integration smoke 真觸發 Softmax@520 確認 onnx 階段就擋；依實測 frontend 耗時評估是否收窄 default-on 粒度。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
from pathlib import Path
|
|
import random
|
|
import shutil
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from services.workers.onnx.core import process_onnx_core
|
|
from services.workers.bie.core import process_bie_core
|
|
from services.workers.nef.core import process_nef_core
|
|
|
|
|
|
def test_worker_flow_e2e_tflite_uses_single_workdir():
|
|
base_outputs = ROOT / "tests" / "fixtures" / "outputs"
|
|
base_outputs.mkdir(parents=True, exist_ok=True)
|
|
|
|
task_id = None
|
|
work_dir = None
|
|
for _ in range(50):
|
|
candidate = random.randint(100, 999)
|
|
candidate_dir = base_outputs / str(candidate)
|
|
if not candidate_dir.exists():
|
|
task_id = candidate
|
|
work_dir = candidate_dir
|
|
break
|
|
assert task_id is not None, "Unable to allocate a unique task id"
|
|
|
|
work_dir.mkdir(parents=True, exist_ok=False)
|
|
|
|
src_onnx_dir = ROOT / "tests" / "fixtures" / "onnx"
|
|
src_images_dir = ROOT / "tests" / "fixtures" / "bie_images"
|
|
ref_images_dir = work_dir / "ref_images"
|
|
|
|
tflite_files = [p for p in src_onnx_dir.iterdir() if p.is_file() and p.suffix == ".tflite"]
|
|
assert len(tflite_files) == 1, "Expected a single TFLite fixture file"
|
|
input_file = work_dir / tflite_files[0].name
|
|
shutil.copy2(tflite_files[0], input_file)
|
|
shutil.copytree(src_images_dir, ref_images_dir)
|
|
|
|
work_inputs = [p for p in work_dir.iterdir() if p.is_file()]
|
|
assert len(work_inputs) == 1, "Working directory must contain a single input file"
|
|
work_input_file = work_inputs[0]
|
|
|
|
onnx_output = work_dir / "out.onnx"
|
|
onnx_params = {
|
|
"model_id": 20,
|
|
"version": "e2e-tflite",
|
|
"platform": "520",
|
|
"work_dir": str(work_dir),
|
|
"enable_evaluate": False,
|
|
# 此測試聚焦端到端流程、非相容性檢查；關掉 pre-check 保持原行為與速度。
|
|
"enable_precheck": False,
|
|
}
|
|
onnx_result = process_onnx_core(
|
|
{"file_path": str(work_input_file)},
|
|
str(onnx_output),
|
|
onnx_params,
|
|
)
|
|
|
|
assert onnx_output.exists()
|
|
assert onnx_result["file_path"] == str(onnx_output)
|
|
assert onnx_result["file_size"] > 0
|
|
|
|
bie_output = work_dir / "out.bie"
|
|
bie_params = {"model_id": 21, "version": "e2e-tflite", "platform": "530", "work_dir": str(work_dir)}
|
|
bie_result = process_bie_core(
|
|
{"onnx_file_path": str(onnx_output), "data_dir": str(ref_images_dir)},
|
|
str(bie_output),
|
|
bie_params,
|
|
)
|
|
|
|
assert bie_output.exists()
|
|
assert bie_result["file_path"] == str(bie_output)
|
|
assert bie_result["file_size"] > 0
|
|
|
|
nef_output = work_dir / "out.nef"
|
|
nef_params = {"model_id": 22, "version": "e2e-tflite", "platform": "730", "work_dir": str(work_dir)}
|
|
nef_result = process_nef_core(
|
|
{"bie_file_path": str(bie_output)},
|
|
str(nef_output),
|
|
nef_params,
|
|
)
|
|
|
|
assert nef_output.exists()
|
|
assert nef_result["file_path"] == str(nef_output)
|
|
assert nef_result["file_size"] > 0
|