jim800121chen
f14d24bd7b
新增 classification 推論支援，核心是「不預設模型長相」： - output shape 自適應：支援 (1,C)/(C,)/(1,C,1,1)/(C,1,1)/(1,1,1,C) 及任何 squeeze 後為一維的張量；類別數從 shape 動態取得， 移除原本寫死的 num_classes=1000 - 無法解析時明確報錯（附實際 shape），不靜默回空結果 - logits vs 已 softmax 自動偵測：sum≈1.0 且全非負才跳過 softmax （容差 1e-4，實測 float32 softmax 偏差最大僅 ~4e-7） - label 為純顯示層：有注入用 label、沒有則輸出原始 enum class_N - input size 三層來源：SDK > 宣告值 > 檔名猜測，log 標示來源 - 新增 handle_set_inference_options：不重載模型即可改解析方式與 label 修正： - _detect_model_type 改為外部指定優先，解決未知檔名被誤判成 tiny_yolov3 而回傳空結果的根因 - handle_disconnect/reset 未清 label 狀態，導致換模型後沿用舊 label 表 - load_model 失敗路徑會污染全域 metadata，改為 defer-until-success - taskType 值域統一為 object_detection（原本回 detection） - 檔名解析原本只取 width 丟棄 height，非正方形模型會被壓成正方形 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1573 lines
71 KiB
Python
1573 lines
71 KiB
Python
#!/usr/bin/env python3
|
||
"""Unit tests for kneron_bridge classification support (M1).
|
||
|
||
Mock-based tests — no real Kneron dongle needed. 覆蓋：
|
||
|
||
- _normalize_task_type：值域統一（detection → object_detection）
|
||
- _sanitize_labels：labels payload 正規化
|
||
- _resolve_label：有/無 label、稀疏、長度不符的 fallback
|
||
- _extract_logits_vector：output shape 自適應（含無法判定時拋錯）
|
||
- _looks_like_probabilities：logits vs 已 softmax 的偵測
|
||
- _parse_classification_output：端到端 post-process
|
||
- _detect_model_type：外部指定優先於檔名猜測（classification 根因修正）
|
||
- handle_load_model / handle_inference：taskType 值域 + detection 不回歸
|
||
|
||
執行方式：
|
||
cd server/scripts && python3 test_kneron_bridge_classification.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
import numpy as np
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
# ── 在 import bridge 前 fake kp module（避免實機相依）─────────────────
|
||
class _FakeChannelOrdering:
|
||
KP_CHANNEL_ORDERING_CHW = "chw"
|
||
|
||
|
||
class _FakeKpInference:
|
||
"""generic_inference_retrieve_float_node 由各 test 用 mock.patch 覆寫。"""
|
||
|
||
def generic_inference_retrieve_float_node(self, **kwargs):
|
||
raise NotImplementedError("must be patched per test")
|
||
|
||
|
||
class _FakeKpCore:
|
||
def disconnect_devices(self, *args, **kwargs):
|
||
return 0
|
||
|
||
|
||
class _FakeKp:
|
||
core = _FakeKpCore()
|
||
inference = _FakeKpInference()
|
||
ChannelOrdering = _FakeChannelOrdering
|
||
|
||
|
||
sys.modules.setdefault("kp", _FakeKp())
|
||
|
||
import kneron_bridge as bridge # noqa: E402
|
||
|
||
|
||
# ── Helpers ──────────────────────────────────────────────────────────
|
||
class FakeOutputNode:
|
||
def __init__(self, ndarray):
|
||
self.ndarray = ndarray
|
||
|
||
|
||
class FakeHeader:
|
||
def __init__(self, num_output_node):
|
||
self.num_output_node = num_output_node
|
||
|
||
|
||
class FakeResult:
|
||
"""Fake generic raw result，帶任意數量 output node。"""
|
||
|
||
def __init__(self, arrays):
|
||
self.header = FakeHeader(len(arrays))
|
||
self._arrays = arrays
|
||
|
||
def retrieve(self, node_idx=0, **kwargs):
|
||
return FakeOutputNode(self._arrays[node_idx])
|
||
|
||
|
||
def patch_retrieve(result):
|
||
"""把 SDK 的 retrieve_float_node 導向 FakeResult。"""
|
||
return mock.patch.object(
|
||
bridge.kp.inference,
|
||
"generic_inference_retrieve_float_node",
|
||
side_effect=lambda **kw: result.retrieve(**kw),
|
||
create=True,
|
||
)
|
||
|
||
|
||
def silence_log():
|
||
return mock.patch.object(bridge, "_log", lambda *a, **k: None)
|
||
|
||
|
||
# models.json 每個 detection model 實際帶的 labels —— 只有前 10 筆 COCO。
|
||
# 生產環境的 detection 路徑拿到的就是這份（不是 None），所以測試必須用它。
|
||
TRUNCATED_COCO_LABELS = bridge.COCO_CLASSES[:10]
|
||
|
||
|
||
def make_yolo_tensor(class_id, grid=7, num_classes=80, num_anchors=3):
|
||
"""Build a Tiny-YOLOv3-shaped tensor with exactly one high-confidence box.
|
||
|
||
Layout: (1, num_anchors * (5 + num_classes), grid, grid)，CHW ordering。
|
||
只在 anchor 0 / cell (0,0) 放一個物件，其類別為 class_id；其餘全部設成
|
||
-10（sigmoid ≈ 0）確保低於 CONF_THRESHOLD 而被濾掉，因此解析結果必定
|
||
是「剛好一個 detection」，斷言才能精確。
|
||
"""
|
||
entry_size = 5 + num_classes
|
||
arr = np.full((1, num_anchors * entry_size, grid, grid), -10.0, dtype=np.float32)
|
||
arr[0, 0, 0, 0] = 0.0 # tx
|
||
arr[0, 1, 0, 0] = 0.0 # ty
|
||
arr[0, 2, 0, 0] = -2.0 # tw（小框，避免 NMS 邊界效應）
|
||
arr[0, 3, 0, 0] = -2.0 # th
|
||
arr[0, 4, 0, 0] = 10.0 # objectness → sigmoid ≈ 1
|
||
arr[0, 5 + class_id, 0, 0] = 10.0 # 該類別分數 → sigmoid ≈ 1
|
||
return arr
|
||
|
||
|
||
class BridgeStateTestCase(unittest.TestCase):
|
||
"""每個 test 前後還原 bridge 的全域狀態，確保測試互相獨立。"""
|
||
|
||
_GLOBALS = ("_model_type", "_model_input_size", "_model_id", "_model_nef",
|
||
"_model_nef_path", "_task_type_override", "_model_labels",
|
||
"_device_group",
|
||
# input size 三層來源（M10 實機驗收修正）。漏還原會讓前一個
|
||
# test 留下的 SDK 來源標記影響後一個 test 的 fallback 判定。
|
||
"_model_input_width", "_model_input_height",
|
||
"_model_input_size_source", "_model_declared_input_size")
|
||
|
||
def setUp(self):
|
||
self._saved = {name: getattr(bridge, name) for name in self._GLOBALS}
|
||
self._log_patch = silence_log()
|
||
self._log_patch.start()
|
||
self.addCleanup(self._log_patch.stop)
|
||
self.addCleanup(self._restore)
|
||
|
||
def _restore(self):
|
||
for name, value in self._saved.items():
|
||
setattr(bridge, name, value)
|
||
|
||
|
||
# ── _normalize_task_type ─────────────────────────────────────────────
|
||
class TestNormalizeTaskType(BridgeStateTestCase):
|
||
def test_classification_passthrough(self):
|
||
self.assertEqual(bridge._normalize_task_type("classification"),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_object_detection_passthrough(self):
|
||
self.assertEqual(bridge._normalize_task_type("object_detection"),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_legacy_detection_alias_maps_to_object_detection(self):
|
||
"""舊值 'detection' 必須映射到統一值域 object_detection（R-4）。"""
|
||
self.assertEqual(bridge._normalize_task_type("detection"),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_case_and_whitespace_insensitive(self):
|
||
self.assertEqual(bridge._normalize_task_type(" Classification "),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_none_and_empty_return_none(self):
|
||
self.assertIsNone(bridge._normalize_task_type(None))
|
||
self.assertIsNone(bridge._normalize_task_type(""))
|
||
self.assertIsNone(bridge._normalize_task_type(" "))
|
||
|
||
def test_unknown_value_returns_none(self):
|
||
self.assertIsNone(bridge._normalize_task_type("segmentation"))
|
||
|
||
def test_non_string_returns_none(self):
|
||
self.assertIsNone(bridge._normalize_task_type(123))
|
||
self.assertIsNone(bridge._normalize_task_type(["classification"]))
|
||
|
||
|
||
# ── _sanitize_labels ─────────────────────────────────────────────────
|
||
class TestSanitizeLabels(BridgeStateTestCase):
|
||
def test_none_returns_none(self):
|
||
self.assertIsNone(bridge._sanitize_labels(None))
|
||
|
||
def test_empty_list_returns_none(self):
|
||
self.assertIsNone(bridge._sanitize_labels([]))
|
||
|
||
def test_all_blank_returns_none(self):
|
||
self.assertIsNone(bridge._sanitize_labels(["", " "]))
|
||
|
||
def test_normal_list_passthrough(self):
|
||
self.assertEqual(bridge._sanitize_labels(["剪刀", "石頭", "布"]),
|
||
["剪刀", "石頭", "布"])
|
||
|
||
def test_none_element_becomes_empty_string(self):
|
||
self.assertEqual(bridge._sanitize_labels(["a", None, "c"]), ["a", "", "c"])
|
||
|
||
def test_non_string_element_coerced(self):
|
||
self.assertEqual(bridge._sanitize_labels(["a", 7]), ["a", "7"])
|
||
|
||
def test_non_list_returns_none(self):
|
||
self.assertIsNone(bridge._sanitize_labels("剪刀,石頭"))
|
||
self.assertIsNone(bridge._sanitize_labels({"0": "a"}))
|
||
|
||
|
||
# ── _resolve_label ───────────────────────────────────────────────────
|
||
class TestResolveLabel(BridgeStateTestCase):
|
||
def test_uses_injected_label(self):
|
||
self.assertEqual(bridge._resolve_label(1, labels=["剪刀", "石頭", "布"]),
|
||
"石頭")
|
||
|
||
def test_no_labels_falls_back_to_enum_index(self):
|
||
"""沒有 label 不是錯誤狀態、輸出原始 enum。"""
|
||
self.assertEqual(bridge._resolve_label(2), "class_2")
|
||
|
||
def test_index_beyond_labels_falls_back_to_enum(self):
|
||
"""labels 長度與類別數不符時，對不到的 fallback 回 index。"""
|
||
self.assertEqual(bridge._resolve_label(5, labels=["a", "b"]), "class_5")
|
||
|
||
def test_sparse_blank_label_falls_back(self):
|
||
self.assertEqual(bridge._resolve_label(1, labels=["a", "", "c"]), "class_1")
|
||
|
||
def test_fallback_labels_used_when_no_injection(self):
|
||
"""detection 沒注入 labels 時沿用 COCO（既有行為不變）。"""
|
||
self.assertEqual(
|
||
bridge._resolve_label(0, labels=None, fallback_labels=bridge.COCO_CLASSES),
|
||
"person")
|
||
|
||
def test_injected_labels_take_priority_over_fallback(self):
|
||
self.assertEqual(
|
||
bridge._resolve_label(0, labels=["自訂"], fallback_labels=bridge.COCO_CLASSES),
|
||
"自訂")
|
||
|
||
def test_negative_index_falls_back(self):
|
||
self.assertEqual(bridge._resolve_label(-1, labels=["a", "b"]), "class_-1")
|
||
|
||
|
||
# ── _extract_logits_vector（output shape 自適應）─────────────────────
|
||
class TestExtractLogitsVector(BridgeStateTestCase):
|
||
def _extract(self, arrays):
|
||
result = FakeResult(arrays)
|
||
with patch_retrieve(result):
|
||
return bridge._extract_logits_vector(result)
|
||
|
||
def test_shape_1xC(self):
|
||
scores, node = self._extract([np.array([[1.0, 2.0, 3.0]])])
|
||
np.testing.assert_allclose(scores, [1.0, 2.0, 3.0])
|
||
self.assertEqual(node, 0)
|
||
self.assertEqual(scores.ndim, 1)
|
||
|
||
def test_shape_C(self):
|
||
scores, _ = self._extract([np.array([4.0, 5.0])])
|
||
np.testing.assert_allclose(scores, [4.0, 5.0])
|
||
|
||
def test_shape_1xCx1x1(self):
|
||
arr = np.array([1.0, 2.0, 3.0]).reshape(1, 3, 1, 1)
|
||
scores, _ = self._extract([arr])
|
||
np.testing.assert_allclose(scores, [1.0, 2.0, 3.0])
|
||
|
||
def test_shape_Cx1x1(self):
|
||
arr = np.array([1.0, 2.0, 3.0]).reshape(3, 1, 1)
|
||
scores, _ = self._extract([arr])
|
||
np.testing.assert_allclose(scores, [1.0, 2.0, 3.0])
|
||
|
||
def test_shape_1x1x1xC_nhwc(self):
|
||
arr = np.array([1.0, 2.0, 3.0]).reshape(1, 1, 1, 3)
|
||
scores, _ = self._extract([arr])
|
||
np.testing.assert_allclose(scores, [1.0, 2.0, 3.0])
|
||
|
||
def test_single_class_scalar_shape(self):
|
||
"""C == 1 squeeze 後是純量，仍須視為合法的一類輸出。"""
|
||
scores, _ = self._extract([np.array([[0.9]])])
|
||
self.assertEqual(scores.shape, (1,))
|
||
np.testing.assert_allclose(scores, [0.9])
|
||
|
||
def test_multiple_nodes_picks_largest_1d(self):
|
||
arrays = [np.array([[0.1, 0.2]]), np.array([[1.0, 2.0, 3.0, 4.0]])]
|
||
scores, node = self._extract(arrays)
|
||
self.assertEqual(node, 1)
|
||
self.assertEqual(scores.size, 4)
|
||
|
||
def test_skips_spatial_nodes_and_uses_1d_node(self):
|
||
"""detection 風格的 (C,H,W) 節點會被跳過、只取一維節點。"""
|
||
arrays = [np.zeros((85, 7, 7)), np.array([[1.0, 2.0, 3.0]])]
|
||
scores, node = self._extract(arrays)
|
||
self.assertEqual(node, 1)
|
||
np.testing.assert_allclose(scores, [1.0, 2.0, 3.0])
|
||
|
||
def test_class_count_is_dynamic_not_hardcoded(self):
|
||
for c in (2, 3, 7, 1000):
|
||
scores, _ = self._extract([np.zeros((1, c))])
|
||
self.assertEqual(scores.size, c)
|
||
|
||
def test_unrecognizable_shape_raises_with_actual_shape(self):
|
||
"""只有 spatial 輸出時必須明確拋錯、且訊息含實際 shape。"""
|
||
with self.assertRaises(ValueError) as ctx:
|
||
self._extract([np.zeros((85, 7, 7))])
|
||
msg = str(ctx.exception)
|
||
self.assertIn("(85, 7, 7)", msg)
|
||
self.assertIn("classification output not recognizable", msg)
|
||
|
||
def test_empty_output_raises(self):
|
||
with self.assertRaises(ValueError):
|
||
self._extract([np.zeros((0,))])
|
||
|
||
def test_error_message_lists_all_nodes(self):
|
||
with self.assertRaises(ValueError) as ctx:
|
||
self._extract([np.zeros((3, 4, 5)), np.zeros((6, 7, 8))])
|
||
msg = str(ctx.exception)
|
||
self.assertIn("node[0]=(3, 4, 5)", msg)
|
||
self.assertIn("node[1]=(6, 7, 8)", msg)
|
||
|
||
|
||
# ── _looks_like_probabilities ────────────────────────────────────────
|
||
class TestLooksLikeProbabilities(BridgeStateTestCase):
|
||
def test_exact_probability_vector_detected(self):
|
||
self.assertTrue(bridge._looks_like_probabilities(np.array([0.9, 0.07, 0.03])))
|
||
|
||
def test_within_tolerance_detected(self):
|
||
self.assertTrue(bridge._looks_like_probabilities(np.array([0.5, 0.50005])))
|
||
|
||
def test_outside_tolerance_not_detected(self):
|
||
self.assertFalse(bridge._looks_like_probabilities(np.array([0.5, 0.6])))
|
||
|
||
def test_real_float32_softmax_still_detected(self):
|
||
"""收緊容差後，真正的 softmax 輸出仍必須被認出（不可誤殺）。
|
||
|
||
float32 softmax 的總和誤差實測最壞約 4e-7，遠小於 1e-4。
|
||
這個測試釘住「容差不可再收到比浮點誤差還小」。
|
||
"""
|
||
for c in (2, 3, 10, 1000):
|
||
logits = np.linspace(-8.0, 8.0, c).astype(np.float32)
|
||
e = np.exp(logits - logits.max())
|
||
probs = (e / e.sum()).astype(np.float32).astype(np.float64)
|
||
self.assertTrue(bridge._looks_like_probabilities(probs),
|
||
msg=f"real softmax output rejected at C={c}")
|
||
|
||
def test_tolerance_is_tight_enough_for_low_class_counts(self):
|
||
"""M-2：容差必須遠小於低類別數 logits 的偶然偏差尺度。"""
|
||
self.assertLessEqual(bridge.PROB_SUM_TOLERANCE, 1e-4)
|
||
|
||
def test_negative_values_are_logits(self):
|
||
self.assertFalse(bridge._looks_like_probabilities(np.array([-1.0, 2.0])))
|
||
|
||
def test_large_logits_not_probabilities(self):
|
||
self.assertFalse(bridge._looks_like_probabilities(np.array([5.0, 3.0, 1.0])))
|
||
|
||
def test_single_class_probability(self):
|
||
self.assertTrue(bridge._looks_like_probabilities(np.array([1.0])))
|
||
|
||
def test_nan_not_probabilities(self):
|
||
self.assertFalse(bridge._looks_like_probabilities(np.array([np.nan, 1.0])))
|
||
|
||
def test_empty_not_probabilities(self):
|
||
self.assertFalse(bridge._looks_like_probabilities(np.array([])))
|
||
|
||
def test_two_class_logits_summing_to_one_is_known_ambiguous(self):
|
||
"""M-2 已知殘留風險：恰好和為 1 的非負 logits 無法與機率區分。
|
||
|
||
[0.4, 0.6] 在數學上與機率向量完全等價，任何只看「非負 + 和為 1」
|
||
的啟發式都會判為機率。此測試不是斷言「這樣是對的」，而是把
|
||
**已知且已接受的行為** 釘住 —— 若未來有人加了第三條件改變此判定，
|
||
測試會失敗、迫使他重新評估是否誤殺真機率。
|
||
"""
|
||
self.assertTrue(bridge._looks_like_probabilities(np.array([0.4, 0.6])))
|
||
|
||
def test_two_class_logits_slightly_off_one_now_caught(self):
|
||
"""收緊容差的實際收益：舊的 0.01 容差會誤判、1e-4 能擋下。"""
|
||
for vec in ([0.4, 0.605], [0.3, 0.695], [0.45, 0.555]):
|
||
arr = np.array(vec)
|
||
# 這些和落在 1.0±0.01 內但超出 1e-4 → 舊實作誤判、新實作正確
|
||
self.assertLess(abs(float(arr.sum()) - 1.0), 0.01)
|
||
self.assertFalse(bridge._looks_like_probabilities(arr),
|
||
msg=f"{vec} should be treated as logits")
|
||
|
||
|
||
# ── _parse_classification_output ─────────────────────────────────────
|
||
class TestParseClassificationOutput(BridgeStateTestCase):
|
||
def _parse(self, arrays, **kwargs):
|
||
result = FakeResult(arrays)
|
||
with patch_retrieve(result):
|
||
return bridge._parse_classification_output(result, **kwargs)
|
||
|
||
def test_logits_get_softmaxed_and_sorted_desc(self):
|
||
out = self._parse([np.array([[1.0, 3.0, 2.0]])])
|
||
self.assertEqual([c["classIndex"] for c in out], [1, 2, 0])
|
||
self.assertAlmostEqual(sum(c["confidence"] for c in out), 1.0, places=6)
|
||
self.assertGreater(out[0]["confidence"], out[1]["confidence"])
|
||
|
||
def test_already_softmaxed_output_is_not_flattened(self):
|
||
"""R-2：已是機率的輸出不可再 softmax（否則 3 類趨近 0.33）。"""
|
||
probs = np.array([[0.94, 0.04, 0.02]])
|
||
out = self._parse([probs])
|
||
self.assertAlmostEqual(out[0]["confidence"], 0.94, places=6)
|
||
self.assertEqual(out[0]["classIndex"], 0)
|
||
|
||
def test_double_softmax_would_have_flattened(self):
|
||
"""對照組：確認若真的再 softmax 一次，top-1 會掉到 ~0.5 以下。"""
|
||
probs = np.array([0.94, 0.04, 0.02])
|
||
double = bridge._softmax(probs)
|
||
self.assertLess(float(np.max(double)), 0.6)
|
||
|
||
def test_labels_replace_display_name(self):
|
||
out = self._parse([np.array([[0.1, 5.0, 0.2]])],
|
||
labels=["剪刀", "石頭", "布"])
|
||
self.assertEqual(out[0]["label"], "石頭")
|
||
self.assertEqual(out[0]["classIndex"], 1)
|
||
|
||
def test_without_labels_uses_raw_enum(self):
|
||
out = self._parse([np.array([[0.1, 5.0, 0.2]])])
|
||
self.assertEqual(out[0]["label"], "class_1")
|
||
self.assertEqual(out[0]["classIndex"], 1)
|
||
|
||
def test_label_count_mismatch_partial_fallback(self):
|
||
"""labels 太短：對到的用 label、對不到的用 index，不整批失敗。"""
|
||
out = self._parse([np.array([[3.0, 2.0, 1.0]])], labels=["甲", "乙"])
|
||
by_index = {c["classIndex"]: c["label"] for c in out}
|
||
self.assertEqual(by_index[0], "甲")
|
||
self.assertEqual(by_index[1], "乙")
|
||
self.assertEqual(by_index[2], "class_2")
|
||
|
||
def test_top_k_default_is_five(self):
|
||
out = self._parse([np.arange(10.0).reshape(1, 10)])
|
||
self.assertEqual(len(out), 5)
|
||
|
||
def test_top_k_capped_by_class_count(self):
|
||
out = self._parse([np.array([[1.0, 2.0, 3.0]])])
|
||
self.assertEqual(len(out), 3)
|
||
|
||
def test_top_k_configurable(self):
|
||
out = self._parse([np.arange(10.0).reshape(1, 10)], top_k=2)
|
||
self.assertEqual(len(out), 2)
|
||
|
||
def test_invalid_top_k_falls_back_to_default(self):
|
||
out = self._parse([np.arange(10.0).reshape(1, 10)], top_k=0)
|
||
self.assertEqual(len(out), bridge.DEFAULT_CLASSIFICATION_TOP_K)
|
||
out = self._parse([np.arange(10.0).reshape(1, 10)], top_k="abc")
|
||
self.assertEqual(len(out), bridge.DEFAULT_CLASSIFICATION_TOP_K)
|
||
|
||
def test_result_schema_fields(self):
|
||
out = self._parse([np.array([[1.0, 2.0]])], labels=["a", "b"])
|
||
self.assertEqual(set(out[0].keys()), {"label", "classIndex", "confidence"})
|
||
self.assertIsInstance(out[0]["label"], str)
|
||
self.assertIsInstance(out[0]["classIndex"], int)
|
||
self.assertIsInstance(out[0]["confidence"], float)
|
||
|
||
def test_unparseable_output_raises_not_silently_empty(self):
|
||
"""絕對不要靜默回傳空結果 —— 必須拋出可讀錯誤。"""
|
||
with self.assertRaises(ValueError):
|
||
self._parse([np.zeros((85, 7, 7))])
|
||
|
||
|
||
# ── _detect_model_type（外部指定優先 = classification 根因修正）──────
|
||
class TestDetectModelType(BridgeStateTestCase):
|
||
def test_unknown_filename_without_hint_falls_back_to_yolo(self):
|
||
"""既有行為：沒有外部指定時仍靠檔名猜（向後相容）。"""
|
||
bridge._detect_model_type(1784536643, "/x/1784536643_models_520.nef")
|
||
self.assertEqual(bridge._model_type, "tiny_yolov3")
|
||
|
||
def test_external_classification_overrides_filename_guess(self):
|
||
"""根因修正：外部指定 classification 就不猜、不再誤判成 tiny_yolov3。"""
|
||
bridge._detect_model_type(1784536643, "/x/1784536643_models_520.nef",
|
||
task_type="classification")
|
||
self.assertEqual(bridge._model_type, "classification")
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_external_classification_on_generic_model_nef(self):
|
||
"""上傳流程存成 model.nef，同樣不含關鍵字。"""
|
||
bridge._detect_model_type(999, "/data/models/uuid/model.nef",
|
||
task_type="classification")
|
||
self.assertEqual(bridge._model_type, "classification")
|
||
|
||
def test_external_object_detection_keeps_detection_backbone(self):
|
||
bridge._detect_model_type(20004, "/x/fcos.nef", task_type="object_detection")
|
||
self.assertEqual(bridge._model_type, "fcos")
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_external_object_detection_overrides_resnet_filename(self):
|
||
"""宣告 detection 但檔名像 resnet → 不可跑 classification 分支。"""
|
||
bridge._detect_model_type(12345, "/x/resnet18_custom.nef",
|
||
task_type="object_detection")
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_known_model_id_still_wins_without_hint(self):
|
||
bridge._detect_model_type(20005, "/x/whatever.nef")
|
||
self.assertEqual(bridge._model_type, "yolov5s")
|
||
self.assertEqual(bridge._model_input_size, 640)
|
||
|
||
def test_classification_hint_still_parses_input_size(self):
|
||
bridge._detect_model_type(None, "/x/custom_w320h320.nef",
|
||
task_type="classification")
|
||
self.assertEqual(bridge._model_type, "classification")
|
||
self.assertEqual(bridge._model_input_size, 320)
|
||
|
||
def test_unknown_task_type_is_ignored_and_falls_back(self):
|
||
bridge._detect_model_type(None, "/x/fcos.nef", task_type="segmentation")
|
||
self.assertEqual(bridge._model_type, "fcos")
|
||
|
||
|
||
# ── Input size 三層來源（SDK > declared > filename guess）─────────────
|
||
#
|
||
# 背景：input size 舊版**完全來自檔名猜測**。使用者的
|
||
# 1784536643_models_520.nef 檔名既無 wNNNhNNN 也無型別關鍵字 → 落 else 分支
|
||
# → 寫死 224 → 圖片被縮到錯的尺寸送進 NPU → 分類結果錯誤，**但不報錯**。
|
||
class FakeTensorDescriptor:
|
||
"""Mirror of kp.TensorDescriptor 的 shape 介面。
|
||
|
||
只實作 bridge 真正會讀的兩個屬性。真 SDK 物件的行為已用 venv 的
|
||
kp.TensorDescriptor 實跑驗證過（見 handover note），此處用 fake 讓
|
||
測試不依賴 KneronPLUS 安裝。
|
||
"""
|
||
|
||
def __init__(self, shape_onnx=None, shape_npu=None):
|
||
self.shape_onnx = list(shape_onnx or [])
|
||
self.shape_npu = list(shape_npu or [])
|
||
|
||
|
||
class FakeSingleModel:
|
||
def __init__(self, model_id=0, input_nodes=None):
|
||
self.id = model_id
|
||
self.input_nodes = list(input_nodes or [])
|
||
|
||
|
||
class FakeNefDescriptor:
|
||
def __init__(self, models=None):
|
||
self.models = list(models or [])
|
||
|
||
|
||
def nef_with_shape(shape_onnx=None, shape_npu=None, model_id=0):
|
||
return FakeNefDescriptor([
|
||
FakeSingleModel(model_id,
|
||
[FakeTensorDescriptor(shape_onnx, shape_npu)])
|
||
])
|
||
|
||
|
||
class TestInputSizeFromSDK(BridgeStateTestCase):
|
||
"""第 1 層：模型自己宣告的 shape，唯一可靠的來源。"""
|
||
|
||
def test_sdk_shape_nchw_wins_over_filename_guess(self):
|
||
"""核心修正：檔名猜不到時不再退回 224，而是問模型自己。"""
|
||
bridge._detect_model_type(
|
||
1784536643, "/x/1784536643_models_520.nef",
|
||
task_type="classification",
|
||
nef=nef_with_shape([1, 3, 320, 320], model_id=1784536643))
|
||
self.assertEqual(bridge._model_input_width, 320)
|
||
self.assertEqual(bridge._model_input_height, 320)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_sdk_wins_over_wrong_declared_value(self):
|
||
"""使用者填 640x640 但模型其實是 320x320 → 必須聽模型的。
|
||
|
||
這正是實機驗收踩到的情境：使用者自己說 640 是「隨便填的」。
|
||
"""
|
||
bridge._detect_model_type(
|
||
1784536643, "/x/1784536643_models_520.nef",
|
||
task_type="classification",
|
||
nef=nef_with_shape([1, 3, 320, 320], model_id=1784536643),
|
||
declared_input_size={"width": 640, "height": 640})
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height),
|
||
(320, 320))
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_sdk_nhwc_shape(self):
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([1, 192, 256, 3],
|
||
model_id=999))
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height),
|
||
(256, 192))
|
||
|
||
def test_sdk_non_square_keeps_both_axes(self):
|
||
"""非正方形不可被壓成正方形 —— 舊版只取 width、height 直接丟掉。"""
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([1, 3, 192, 256],
|
||
model_id=999))
|
||
self.assertEqual(bridge._model_input_width, 256)
|
||
self.assertEqual(bridge._model_input_height, 192)
|
||
|
||
def test_scalar_input_size_is_the_shorter_edge(self):
|
||
"""派生純量取短邊：它的用途是 min_dim，取長邊會讓短軸不足。"""
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([1, 3, 192, 256],
|
||
model_id=999))
|
||
self.assertEqual(bridge._model_input_size, 192)
|
||
|
||
def test_shape_npu_used_when_shape_onnx_empty(self):
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([], [1, 3, 512, 512],
|
||
model_id=999))
|
||
self.assertEqual(bridge._model_input_width, 512)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_picks_node_matching_model_id_in_multi_model_nef(self):
|
||
"""一個 .nef 可包多個 model，要取 _model_id 指的那個。"""
|
||
nef = FakeNefDescriptor([
|
||
FakeSingleModel(111, [FakeTensorDescriptor([1, 3, 128, 128])]),
|
||
FakeSingleModel(222, [FakeTensorDescriptor([1, 3, 640, 640])]),
|
||
])
|
||
bridge._detect_model_type(222, "/x/m.nef", nef=nef)
|
||
self.assertEqual(bridge._model_input_width, 640)
|
||
|
||
def test_uninterpretable_shape_falls_through_to_declared(self):
|
||
"""判讀不了就 fallback，**不猜** —— 猜錯就是用錯尺寸推論。"""
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([1, 1000], model_id=999),
|
||
declared_input_size={"width": 300,
|
||
"height": 300})
|
||
self.assertEqual(bridge._model_input_width, 300)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_DECLARED)
|
||
|
||
def test_empty_nef_falls_through_to_filename(self):
|
||
bridge._detect_model_type(
|
||
20004, "/x/kl520_20004_fcos-drk53s_w512h512.nef",
|
||
nef=FakeNefDescriptor([]))
|
||
self.assertEqual(bridge._model_input_width, 512)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
|
||
def test_broken_nef_object_does_not_raise(self):
|
||
"""SDK 物件壞掉不可讓 load_model 整個炸掉，降級即可。"""
|
||
class Exploding:
|
||
@property
|
||
def models(self):
|
||
raise RuntimeError("boom")
|
||
|
||
bridge._detect_model_type(20004, "/x/fcos.nef", nef=Exploding())
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
self.assertEqual(bridge._model_input_width, 512)
|
||
|
||
|
||
class TestInputSizeDeclared(BridgeStateTestCase):
|
||
"""第 2 層：models.json / metadata.json 宣告值。人填的，可能亂填。"""
|
||
|
||
def test_declared_used_when_no_sdk(self):
|
||
bridge._detect_model_type(1784536643, "/x/1784536643_models_520.nef",
|
||
task_type="classification",
|
||
declared_input_size={"width": 224,
|
||
"height": 224})
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_DECLARED)
|
||
|
||
def test_declared_beats_filename_guess(self):
|
||
bridge._detect_model_type(None, "/x/custom_w320h320.nef",
|
||
declared_input_size={"width": 416,
|
||
"height": 416})
|
||
self.assertEqual(bridge._model_input_width, 416)
|
||
|
||
def test_declared_accepts_tuple(self):
|
||
self.assertEqual(bridge._normalize_declared_input_size((256, 192)),
|
||
(256, 192))
|
||
|
||
def test_declared_non_square(self):
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
declared_input_size={"width": 256,
|
||
"height": 192})
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height),
|
||
(256, 192))
|
||
|
||
def test_declared_zero_is_rejected(self):
|
||
"""models.json 沒填 inputSize 時 Go 端會送 0 —— 不可當成有效值。"""
|
||
self.assertIsNone(
|
||
bridge._normalize_declared_input_size({"width": 0, "height": 0}))
|
||
|
||
def test_declared_absurd_value_is_rejected(self):
|
||
self.assertIsNone(
|
||
bridge._normalize_declared_input_size({"width": 999999,
|
||
"height": 999999}))
|
||
|
||
def test_declared_partially_invalid_is_rejected_entirely(self):
|
||
"""半套採用比整組不用更危險 —— 看起來像有正確來源。"""
|
||
self.assertIsNone(
|
||
bridge._normalize_declared_input_size({"width": 320, "height": 0}))
|
||
|
||
def test_declared_non_numeric_is_rejected(self):
|
||
self.assertIsNone(
|
||
bridge._normalize_declared_input_size({"width": "abc",
|
||
"height": "abc"}))
|
||
|
||
def test_declared_none_is_rejected(self):
|
||
self.assertIsNone(bridge._normalize_declared_input_size(None))
|
||
|
||
def test_rejected_declared_falls_back_to_filename(self):
|
||
bridge._detect_model_type(
|
||
20004, "/x/kl520_20004_fcos-drk53s_w512h512.nef",
|
||
declared_input_size={"width": 0, "height": 0})
|
||
self.assertEqual(bridge._model_input_width, 512)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
|
||
|
||
class TestInputSizeFilenameFallback(BridgeStateTestCase):
|
||
"""第 3 層：檔名猜測。維持既有行為，不可回歸。"""
|
||
|
||
def test_bundled_models_resolve_exactly_as_before(self):
|
||
"""既有 detection 模型的尺寸一個都不能變。
|
||
|
||
這些檔名帶 wNNNhNNN / 已知 model id，舊版猜對了；新版三層 fallback
|
||
在沒有 SDK / declared 時必須落到同樣的值。
|
||
"""
|
||
expected = [
|
||
(0, "kl520_tiny_yolo_v3.nef", "tiny_yolov3", 224),
|
||
(20001, "kl520_20001_resnet18_w224h224.nef", "resnet18", 224),
|
||
(20004, "kl520_20004_fcos-drk53s_w512h512.nef", "fcos", 512),
|
||
(20005, "kl520_20005_yolov5-noupsample_w640h640.nef", "yolov5s", 640),
|
||
(None, "kl520_ssd_fd_lm.nef", "ssd", 320),
|
||
(20001, "kl720_20001_resnet18_w224h224.nef", "resnet18", 224),
|
||
(20004, "kl720_20004_fcos-drk53s_w512h512.nef", "fcos", 512),
|
||
(20005, "kl720_20005_yolov5-noupsample_w640h640.nef", "yolov5s", 640),
|
||
]
|
||
for model_id, name, want_type, want_size in expected:
|
||
with self.subTest(nef=name):
|
||
bridge._detect_model_type(model_id, "/data/nef/" + name)
|
||
self.assertEqual(bridge._model_type, want_type)
|
||
self.assertEqual(bridge._model_input_size, want_size)
|
||
self.assertEqual(bridge._model_input_width, want_size)
|
||
self.assertEqual(bridge._model_input_height, want_size)
|
||
|
||
def test_filename_non_square_keeps_height(self):
|
||
"""舊版 _parse_size_from_name 只取 width、丟掉 height。"""
|
||
self.assertEqual(bridge._parse_size_from_name("m_w256h192.nef"),
|
||
(256, 192))
|
||
|
||
def test_filename_without_size_uses_default_for_both_axes(self):
|
||
self.assertEqual(bridge._parse_size_from_name("m.nef", default=512),
|
||
(512, 512))
|
||
|
||
def test_absurd_filename_size_is_rejected(self):
|
||
self.assertEqual(
|
||
bridge._parse_size_from_name("m_w99999999h99999999.nef",
|
||
default=224),
|
||
(224, 224))
|
||
|
||
def test_source_marked_unreliable_in_log_description(self):
|
||
"""實機驗收要能一眼看出尺寸是猜的。"""
|
||
bridge._detect_model_type(1784536643, "/x/1784536643_models_520.nef")
|
||
self.assertIn("UNRELIABLE", bridge._describe_input_size())
|
||
|
||
def test_sdk_source_not_marked_unreliable(self):
|
||
bridge._detect_model_type(999, "/x/m.nef",
|
||
nef=nef_with_shape([1, 3, 320, 320],
|
||
model_id=999))
|
||
desc = bridge._describe_input_size()
|
||
self.assertIn("source: SDK", desc)
|
||
self.assertNotIn("UNRELIABLE", desc)
|
||
|
||
|
||
class TestInputSizeSourceIsolation(BridgeStateTestCase):
|
||
"""來源標記不可跨模型殘留。"""
|
||
|
||
def test_previous_sdk_source_does_not_block_next_model(self):
|
||
"""先載一個有 SDK shape 的，再載一個沒有的 → 後者不可沿用前者尺寸。
|
||
|
||
少了 _resolve_input_size 開頭的降級，第二次會因為「已有更可信來源」
|
||
而不敢寫，靜默沿用 320 —— 只在換模型時才出現，極難 debug。
|
||
"""
|
||
bridge._detect_model_type(999, "/x/a.nef",
|
||
nef=nef_with_shape([1, 3, 320, 320],
|
||
model_id=999))
|
||
self.assertEqual(bridge._model_input_width, 320)
|
||
|
||
bridge._detect_model_type(
|
||
20004, "/x/kl520_20004_fcos-drk53s_w512h512.nef")
|
||
self.assertEqual(bridge._model_input_width, 512)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
|
||
def test_reset_model_metadata_restores_defaults(self):
|
||
bridge._detect_model_type(999, "/x/a.nef",
|
||
nef=nef_with_shape([1, 3, 640, 480],
|
||
model_id=999))
|
||
bridge._reset_model_metadata()
|
||
self.assertEqual(bridge._model_input_width, 224)
|
||
self.assertEqual(bridge._model_input_height, 224)
|
||
self.assertEqual(bridge._model_input_size, 224)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_DEFAULT)
|
||
self.assertIsNone(bridge._model_declared_input_size)
|
||
|
||
|
||
# ── _current_task_type ───────────────────────────────────────────────
|
||
class TestCurrentTaskType(BridgeStateTestCase):
|
||
def test_legacy_resnet18_maps_to_classification(self):
|
||
bridge._model_type = "resnet18"
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_classification_maps_to_classification(self):
|
||
bridge._model_type = "classification"
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_detection_types_map_to_object_detection(self):
|
||
for t in ("tiny_yolov3", "yolov5s", "fcos", "ssd"):
|
||
bridge._model_type = t
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION,
|
||
msg=f"model_type={t}")
|
||
|
||
def test_never_returns_legacy_detection_string(self):
|
||
"""R-4：值域統一，不可再回傳 'detection'。"""
|
||
for t in ("tiny_yolov3", "classification", "unknown_type"):
|
||
bridge._model_type = t
|
||
self.assertNotEqual(bridge._current_task_type(), "detection")
|
||
|
||
|
||
# ── _reset_model_metadata ────────────────────────────────────────────
|
||
class TestResetModelMetadata(BridgeStateTestCase):
|
||
def test_clears_injected_metadata(self):
|
||
bridge._task_type_override = bridge.TASK_TYPE_CLASSIFICATION
|
||
bridge._model_labels = ["a", "b"]
|
||
bridge._reset_model_metadata()
|
||
self.assertIsNone(bridge._task_type_override)
|
||
self.assertIsNone(bridge._model_labels)
|
||
|
||
|
||
# ── handle_load_model ────────────────────────────────────────────────
|
||
class TestHandleLoadModel(BridgeStateTestCase):
|
||
def setUp(self):
|
||
super().setUp()
|
||
bridge._device_group = object()
|
||
|
||
class FakeModel:
|
||
id = 1784536643
|
||
|
||
class FakeNef:
|
||
models = [FakeModel()]
|
||
target_chip = "KL520"
|
||
|
||
self._fake_nef = FakeNef()
|
||
self._load_patch = mock.patch.object(
|
||
bridge.kp.core, "load_model_from_file",
|
||
side_effect=lambda **kw: self._fake_nef, create=True)
|
||
self._load_patch.start()
|
||
self.addCleanup(self._load_patch.stop)
|
||
self._exists_patch = mock.patch.object(os.path, "exists", lambda p: True)
|
||
self._exists_patch.start()
|
||
self.addCleanup(self._exists_patch.stop)
|
||
|
||
def test_classification_hint_is_honored(self):
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/1784536643_models_520.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_CLASSIFICATION)
|
||
self.assertEqual(res["model_type"], "classification")
|
||
self.assertEqual(res["label_count"], 3)
|
||
self.assertEqual(bridge._model_labels, ["剪刀", "石頭", "布"])
|
||
|
||
def test_without_hint_behaviour_unchanged(self):
|
||
"""向後相容：不帶新欄位時與改動前一致（走檔名猜測）。"""
|
||
res = bridge.handle_load_model({"path": "/x/1784536643_models_520.nef"})
|
||
self.assertEqual(res["model_type"], "tiny_yolov3")
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
self.assertEqual(res["label_count"], 0)
|
||
self.assertIsNone(bridge._model_labels)
|
||
|
||
def test_legacy_response_fields_preserved(self):
|
||
res = bridge.handle_load_model({"path": "/x/m.nef"})
|
||
for key in ("status", "model_id", "model_type", "input_size",
|
||
"model_path", "target_chip"):
|
||
self.assertIn(key, res)
|
||
self.assertEqual(res["status"], "loaded")
|
||
|
||
def test_labels_without_task_type_still_stored(self):
|
||
res = bridge.handle_load_model({"path": "/x/fcos.nef",
|
||
"labels": ["自訂A", "自訂B"]})
|
||
self.assertEqual(res["label_count"], 2)
|
||
self.assertEqual(bridge._model_labels, ["自訂A", "自訂B"])
|
||
|
||
def test_reload_clears_previous_labels(self):
|
||
bridge.handle_load_model({"path": "/x/a.nef", "labels": ["舊"]})
|
||
bridge.handle_load_model({"path": "/x/b.nef"})
|
||
self.assertIsNone(bridge._model_labels)
|
||
|
||
# ── input size 三層來源的端到端接線 ──────────────────────────────
|
||
def test_input_size_taken_from_sdk_when_nef_reports_shape(self):
|
||
"""端到端：SDK 有 shape 就用它，不再靠檔名猜。
|
||
|
||
使用者的 .nef 檔名不含尺寸資訊，舊版必定落到寫死的 224。
|
||
"""
|
||
self._fake_nef = nef_with_shape([1, 3, 320, 320], model_id=1784536643)
|
||
self._fake_nef.target_chip = "KL520"
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/1784536643_models_520.nef",
|
||
"task_type": "classification",
|
||
})
|
||
self.assertEqual(res["input_width"], 320)
|
||
self.assertEqual(res["input_height"], 320)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_sdk_overrides_wrong_declared_input_size(self):
|
||
"""實機驗收情境：使用者自承 640 是隨便填的，模型其實是 320。"""
|
||
self._fake_nef = nef_with_shape([1, 3, 320, 320], model_id=1784536643)
|
||
self._fake_nef.target_chip = "KL520"
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/1784536643_models_520.nef",
|
||
"task_type": "classification",
|
||
"input_size": {"width": 640, "height": 640},
|
||
})
|
||
self.assertEqual(res["input_width"], 320)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_declared_input_size_used_when_sdk_silent(self):
|
||
"""SDK 沒回報 shape（FakeNef 無 input_nodes）→ 用宣告值，不是 224。"""
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/1784536643_models_520.nef",
|
||
"task_type": "classification",
|
||
"input_size": {"width": 320, "height": 320},
|
||
})
|
||
self.assertEqual(res["input_width"], 320)
|
||
self.assertEqual(res["input_size"], 320)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_DECLARED)
|
||
|
||
def test_filename_guess_when_neither_sdk_nor_declared(self):
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/kl520_20004_fcos-drk53s_w512h512.nef"})
|
||
self.assertEqual(res["input_size"], 512)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
|
||
def test_input_size_fields_absent_from_request_is_backward_compatible(self):
|
||
"""不帶 input_size 的舊呼叫端行為完全不變。"""
|
||
res = bridge.handle_load_model({"path": "/x/1784536643_models_520.nef"})
|
||
self.assertEqual(res["input_size"], 224)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_FILENAME)
|
||
|
||
def test_failed_load_does_not_change_input_size(self):
|
||
"""defer-until-success：載入失敗 → 尺寸與來源都不可被動到。"""
|
||
bridge.handle_load_model({
|
||
"path": "/x/kl520_20004_fcos-drk53s_w512h512.nef"})
|
||
before = (bridge._model_input_width, bridge._model_input_height,
|
||
bridge._model_input_size_source,
|
||
bridge._model_declared_input_size)
|
||
|
||
with mock.patch.object(bridge.kp.core, "load_model_from_file",
|
||
side_effect=RuntimeError("boom"), create=True):
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/other.nef",
|
||
"input_size": {"width": 640, "height": 640}})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertEqual(
|
||
(bridge._model_input_width, bridge._model_input_height,
|
||
bridge._model_input_size_source,
|
||
bridge._model_declared_input_size),
|
||
before)
|
||
|
||
def test_missing_file_returns_error(self):
|
||
self._exists_patch.stop()
|
||
with mock.patch.object(os.path, "exists", lambda p: False):
|
||
res = bridge.handle_load_model({"path": "/nope.nef"})
|
||
self.assertIn("error", res)
|
||
self._exists_patch.start()
|
||
|
||
def test_no_device_returns_error(self):
|
||
bridge._device_group = None
|
||
res = bridge.handle_load_model({"path": "/x/m.nef"})
|
||
self.assertEqual(res["error"], "device not connected")
|
||
|
||
# ── M-1：失敗路徑不可污染全域 metadata ──────────────────────────
|
||
#
|
||
# 不變式：_model_labels / _task_type_override 必須永遠描述
|
||
# 「_model_id 當前指向的模型」。載入失敗時舊模型仍在裝置上、仍可被
|
||
# 推論，若 labels 已被新模型的值覆蓋 → 舊模型配新 label 表、標籤全錯。
|
||
|
||
def _load_first_model(self):
|
||
"""先成功載入一個 detection 模型，作為「舊模型」基準。"""
|
||
bridge.handle_load_model({
|
||
"path": "/x/fcos.nef",
|
||
"task_type": "object_detection",
|
||
"labels": ["舊A", "舊B"],
|
||
})
|
||
return {
|
||
"model_id": bridge._model_id,
|
||
"model_type": bridge._model_type,
|
||
"labels": list(bridge._model_labels),
|
||
"task_type_override": bridge._task_type_override,
|
||
}
|
||
|
||
def _assert_state_unchanged(self, before):
|
||
self.assertEqual(bridge._model_id, before["model_id"])
|
||
self.assertEqual(bridge._model_type, before["model_type"])
|
||
self.assertEqual(bridge._model_labels, before["labels"])
|
||
self.assertEqual(bridge._task_type_override, before["task_type_override"])
|
||
|
||
def test_load_failure_does_not_pollute_labels(self):
|
||
"""load_model_from_file 拋錯 → labels 不可被新模型的值覆蓋。"""
|
||
before = self._load_first_model()
|
||
|
||
with mock.patch.object(bridge.kp.core, "load_model_from_file",
|
||
side_effect=RuntimeError("error 40"),
|
||
create=True):
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/new_classification.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertIn("error 40", res["error"])
|
||
self._assert_state_unchanged(before)
|
||
|
||
def test_load_failure_keeps_task_type_consistent_with_model_id(self):
|
||
"""失敗後 _current_task_type() 必須仍描述舊模型（detection）。"""
|
||
self._load_first_model()
|
||
|
||
with mock.patch.object(bridge.kp.core, "load_model_from_file",
|
||
side_effect=RuntimeError("boom"), create=True):
|
||
bridge.handle_load_model({
|
||
"path": "/x/c.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
# 若 _task_type_override 被污染成 classification，後續推論會走
|
||
# classification 分支解析 detection 模型的輸出 → 直接拋錯。
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_models_list_failure_does_not_pollute_state(self):
|
||
"""取 nef.models[0] 失敗（空 list）→ 同樣不可留下半套狀態。"""
|
||
before = self._load_first_model()
|
||
|
||
class EmptyNef:
|
||
models = []
|
||
target_chip = "KL520"
|
||
|
||
with mock.patch.object(bridge.kp.core, "load_model_from_file",
|
||
side_effect=lambda **kw: EmptyNef(), create=True):
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/broken.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
self.assertIn("error", res)
|
||
self._assert_state_unchanged(before)
|
||
# _model_nef 也不可被換成載入失敗的那個 nef
|
||
self.assertNotIsInstance(bridge._model_nef, EmptyNef)
|
||
|
||
def test_missing_file_does_not_pollute_state(self):
|
||
"""更早的 early return（檔案不存在）同樣不可動全域。"""
|
||
before = self._load_first_model()
|
||
|
||
self._exists_patch.stop()
|
||
try:
|
||
with mock.patch.object(os.path, "exists", lambda p: False):
|
||
res = bridge.handle_load_model({
|
||
"path": "/nope.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
finally:
|
||
self._exists_patch.start()
|
||
|
||
self.assertIn("error", res)
|
||
self._assert_state_unchanged(before)
|
||
|
||
def test_device_not_connected_does_not_pollute_state(self):
|
||
before = self._load_first_model()
|
||
|
||
bridge._device_group = None
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/c.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
self.assertIn("error", res)
|
||
self._assert_state_unchanged(before)
|
||
|
||
def test_successful_load_still_commits_new_metadata(self):
|
||
"""對照組：成功路徑必須真的換成新 metadata（不是永遠不寫）。"""
|
||
self._load_first_model()
|
||
|
||
res = bridge.handle_load_model({
|
||
"path": "/x/1784536643_models_520.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
self.assertEqual(res["status"], "loaded")
|
||
self.assertEqual(bridge._model_labels, ["剪刀", "石頭", "布"])
|
||
self.assertEqual(bridge._task_type_override,
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
|
||
# ── handle_set_inference_options（M4：推論期切換解析方式 / label）─────
|
||
class TestHandleSetInferenceOptions(BridgeStateTestCase):
|
||
"""推論期動態切換：不重新 load_model 就換解析方式與 label 表。
|
||
|
||
這個 handler 的每一條 early return 都必須是 all-or-nothing —— 半套寫入
|
||
會產生「新解析方式配舊 label 表」這種不報錯的錯誤，跟 M-1 修的
|
||
load_model 污染問題是同一類。
|
||
"""
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
bridge._device_group = object()
|
||
|
||
class FakeModel:
|
||
id = 1784536643
|
||
|
||
class FakeNef:
|
||
models = [FakeModel()]
|
||
target_chip = "KL520"
|
||
|
||
self._fake_nef = FakeNef()
|
||
self._load_patch = mock.patch.object(
|
||
bridge.kp.core, "load_model_from_file",
|
||
side_effect=lambda **kw: self._fake_nef, create=True)
|
||
self._load_patch.start()
|
||
self.addCleanup(self._load_patch.stop)
|
||
self._exists_patch = mock.patch.object(os.path, "exists", lambda p: True)
|
||
self._exists_patch.start()
|
||
self.addCleanup(self._exists_patch.stop)
|
||
|
||
def _load_detection_model(self, path="/x/fcos_w512h512.nef"):
|
||
"""先載一個 detection 模型作為「當前已載入」的基準。"""
|
||
bridge.handle_load_model({
|
||
"path": path,
|
||
"task_type": "object_detection",
|
||
"labels": ["舊A", "舊B"],
|
||
})
|
||
|
||
def _snapshot(self):
|
||
return {
|
||
"model_id": bridge._model_id,
|
||
"model_type": bridge._model_type,
|
||
"input_size": bridge._model_input_size,
|
||
"labels": None if bridge._model_labels is None else list(bridge._model_labels),
|
||
"task_type_override": bridge._task_type_override,
|
||
}
|
||
|
||
def _assert_unchanged(self, before):
|
||
self.assertEqual(bridge._model_id, before["model_id"])
|
||
self.assertEqual(bridge._model_type, before["model_type"])
|
||
self.assertEqual(bridge._model_input_size, before["input_size"])
|
||
self.assertEqual(bridge._model_labels, before["labels"])
|
||
self.assertEqual(bridge._task_type_override, before["task_type_override"])
|
||
|
||
# ── 核心：切換解析方式 ────────────────────────────────────────
|
||
def test_switch_detection_to_classification(self):
|
||
self._load_detection_model()
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertEqual(res["status"], "updated")
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_CLASSIFICATION)
|
||
self.assertEqual(bridge._model_type, "classification")
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_switch_classification_back_to_detection(self):
|
||
bridge.handle_load_model({"path": "/x/m.nef", "task_type": "classification"})
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "object_detection"})
|
||
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_legacy_detection_alias_accepted(self):
|
||
"""bridge 層仍收 'detection' 別名（Go 端另有更嚴的值域把關）。"""
|
||
self._load_detection_model()
|
||
res = bridge.handle_set_inference_options({"task_type": "detection"})
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_switch_does_not_change_input_size(self):
|
||
"""切解析方式不應該偷改 input size（靠保留的 .nef 路徑重推）。"""
|
||
self._load_detection_model(path="/x/fcos_w512h512.nef")
|
||
before_size = bridge._model_input_size
|
||
self.assertEqual(before_size, 512, "前置條件：檔名應解析出 512")
|
||
|
||
bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertEqual(bridge._model_input_size, before_size)
|
||
|
||
def test_switch_preserves_sdk_input_size(self):
|
||
"""切解析方式不可讓 SDK 尺寸退回檔名猜測。
|
||
|
||
_detect_model_type 會被重跑，若沒把 _model_nef 一起帶回去，尺寸就會
|
||
從模型宣告的真值掉回檔名猜的值 —— 而且不報錯。
|
||
"""
|
||
self._fake_nef = nef_with_shape([1, 3, 320, 320], model_id=1784536643)
|
||
self._fake_nef.target_chip = "KL520"
|
||
bridge.handle_load_model({"path": "/x/fcos_w512h512.nef",
|
||
"task_type": "object_detection"})
|
||
self.assertEqual(bridge._model_input_width, 320)
|
||
self.assertEqual(bridge._model_input_size_source,
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertEqual(bridge._model_input_width, 320)
|
||
self.assertEqual(bridge._model_input_height, 320)
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_SDK)
|
||
|
||
def test_switch_preserves_declared_input_size(self):
|
||
"""同理：declared 也不可在切換時被檔名猜測蓋掉。"""
|
||
bridge.handle_load_model({"path": "/x/fcos_w512h512.nef",
|
||
"task_type": "object_detection",
|
||
"input_size": {"width": 320, "height": 256}})
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height),
|
||
(320, 256))
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height),
|
||
(320, 256))
|
||
self.assertEqual(res["input_size_source"],
|
||
bridge.INPUT_SIZE_SOURCE_DECLARED)
|
||
|
||
def test_labels_only_change_does_not_touch_input_size(self):
|
||
self._fake_nef = nef_with_shape([1, 3, 320, 320], model_id=1784536643)
|
||
self._fake_nef.target_chip = "KL520"
|
||
bridge.handle_load_model({"path": "/x/fcos_w512h512.nef",
|
||
"task_type": "classification"})
|
||
before = (bridge._model_input_width, bridge._model_input_height,
|
||
bridge._model_input_size_source)
|
||
|
||
bridge.handle_set_inference_options({"labels": ["a", "b"]})
|
||
|
||
self.assertEqual((bridge._model_input_width, bridge._model_input_height,
|
||
bridge._model_input_size_source), before)
|
||
|
||
# ── 核心：切換 label 表 ───────────────────────────────────────
|
||
def test_set_labels_only(self):
|
||
self._load_detection_model()
|
||
res = bridge.handle_set_inference_options({"labels": ["剪刀", "石頭", "布"]})
|
||
|
||
self.assertEqual(res["label_count"], 3)
|
||
self.assertEqual(bridge._model_labels, ["剪刀", "石頭", "布"])
|
||
# 只給 labels 不應動到解析方式
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
|
||
def test_set_both_task_type_and_labels(self):
|
||
self._load_detection_model()
|
||
res = bridge.handle_set_inference_options({
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
self.assertEqual(res["task_type"], bridge.TASK_TYPE_CLASSIFICATION)
|
||
self.assertEqual(bridge._model_labels, ["剪刀", "石頭", "布"])
|
||
|
||
def test_empty_labels_clears_table(self):
|
||
"""空 list = 清掉 label 表、回到原始 enum（class_N）。刻意可達。"""
|
||
bridge.handle_load_model({"path": "/x/m.nef",
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"]})
|
||
res = bridge.handle_set_inference_options({"labels": []})
|
||
|
||
self.assertEqual(res["label_count"], 0)
|
||
self.assertIsNone(bridge._model_labels)
|
||
self.assertEqual(bridge._resolve_label(1), "class_1")
|
||
|
||
def test_null_labels_clears_table(self):
|
||
bridge.handle_load_model({"path": "/x/m.nef", "labels": ["a", "b"]})
|
||
bridge.handle_set_inference_options({"labels": None})
|
||
self.assertIsNone(bridge._model_labels)
|
||
|
||
def test_sparse_labels_preserved(self):
|
||
"""稀疏 label（空字串佔位）必須原樣保留，由 _resolve_label 決定 fallback。"""
|
||
self._load_detection_model()
|
||
bridge.handle_set_inference_options({"labels": ["a", "", "c"]})
|
||
self.assertEqual(bridge._model_labels, ["a", "", "c"])
|
||
self.assertEqual(bridge._resolve_label(1, labels=bridge._model_labels),
|
||
"class_1")
|
||
|
||
def test_task_type_only_keeps_existing_labels(self):
|
||
"""沒帶 labels 欄位 → label 表原封不動（不是被清空）。"""
|
||
bridge.handle_load_model({"path": "/x/m.nef", "labels": ["保留A", "保留B"]})
|
||
bridge.handle_set_inference_options({"task_type": "classification"})
|
||
self.assertEqual(bridge._model_labels, ["保留A", "保留B"])
|
||
|
||
# ── 前置條件守衛 ─────────────────────────────────────────────
|
||
def test_no_device_returns_error(self):
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
bridge._device_group = None
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertEqual(res["error"], "device not connected")
|
||
self._assert_unchanged(before)
|
||
|
||
def test_no_model_loaded_returns_error(self):
|
||
"""裝置連了但沒載 model → 明確報錯，不能靜默接受設定。"""
|
||
bridge._model_id = None
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertIn("no model loaded", res["error"])
|
||
|
||
def test_empty_params_returns_error(self):
|
||
"""兩個欄位都沒帶 = 呼叫端搞錯，回 200 等於假裝做了事。"""
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
|
||
res = bridge.handle_set_inference_options({})
|
||
|
||
self.assertIn("error", res)
|
||
self._assert_unchanged(before)
|
||
|
||
# ── 驗證失敗 → 零全域變動（defer-until-success）────────────────
|
||
def test_invalid_task_type_rejected_and_state_untouched(self):
|
||
"""與 load_model 不同：這裡不可 fallback 猜測，必須明確報錯。
|
||
|
||
靜默忽略會回 200 但什麼都沒切換，使用者以為生效了 —— 正是這個功能
|
||
要防的靜默失敗。
|
||
"""
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
|
||
res = bridge.handle_set_inference_options({"task_type": "segmentation"})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertIn("invalid task_type", res["error"])
|
||
self._assert_unchanged(before)
|
||
|
||
def test_invalid_task_type_does_not_apply_labels(self):
|
||
"""關鍵：task_type 非法時，同一次請求帶的 labels 也不可被寫入。
|
||
|
||
若先寫 labels 再驗 task_type，就會出現「舊解析方式 + 新 label 表」，
|
||
跟 M-1 修掉的污染是同一類的靜默錯誤。
|
||
"""
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
|
||
res = bridge.handle_set_inference_options({
|
||
"task_type": "bogus",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertEqual(bridge._model_labels, before["labels"],
|
||
"task_type 驗證失敗時 labels 不可被寫入")
|
||
self._assert_unchanged(before)
|
||
|
||
def test_non_list_labels_rejected_and_state_untouched(self):
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
|
||
res = bridge.handle_set_inference_options({"labels": "剪刀,石頭,布"})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertIn("must be a list", res["error"])
|
||
self._assert_unchanged(before)
|
||
|
||
def test_non_list_labels_does_not_apply_task_type(self):
|
||
"""反向：labels 非法時，同一次請求帶的 task_type 也不可生效。"""
|
||
self._load_detection_model()
|
||
before = self._snapshot()
|
||
|
||
res = bridge.handle_set_inference_options({
|
||
"task_type": "classification",
|
||
"labels": {"0": "剪刀"},
|
||
})
|
||
|
||
self.assertIn("error", res)
|
||
self.assertEqual(bridge._task_type_override, before["task_type_override"],
|
||
"labels 驗證失敗時 task_type 不可被寫入")
|
||
self._assert_unchanged(before)
|
||
|
||
# ── 生命週期 ─────────────────────────────────────────────────
|
||
def test_options_cleared_on_disconnect(self):
|
||
self._load_detection_model()
|
||
bridge.handle_set_inference_options({
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
bridge.handle_disconnect({})
|
||
|
||
self.assertIsNone(bridge._model_labels)
|
||
self.assertIsNone(bridge._task_type_override)
|
||
self.assertEqual(bridge._model_nef_path, "")
|
||
|
||
def test_reload_model_overrides_runtime_options(self):
|
||
"""重新 load_model 是 source of truth，會蓋掉推論期的臨時設定。"""
|
||
self._load_detection_model()
|
||
bridge.handle_set_inference_options({
|
||
"task_type": "classification",
|
||
"labels": ["剪刀", "石頭", "布"],
|
||
})
|
||
|
||
bridge.handle_load_model({"path": "/x/fcos.nef",
|
||
"task_type": "object_detection"})
|
||
|
||
self.assertEqual(bridge._current_task_type(),
|
||
bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
self.assertIsNone(bridge._model_labels)
|
||
|
||
def test_response_shape(self):
|
||
self._load_detection_model()
|
||
res = bridge.handle_set_inference_options({"task_type": "classification"})
|
||
for key in ("status", "model_id", "model_type", "task_type",
|
||
"label_count", "input_size"):
|
||
self.assertIn(key, res)
|
||
|
||
def test_dispatch_registered_in_main_loop(self):
|
||
"""指令要真的接得到 —— handler 寫好但沒掛進 dispatch 是靜默失效。"""
|
||
import ast
|
||
import pathlib
|
||
src = pathlib.Path(bridge.__file__).read_text(encoding="utf-8")
|
||
tree = ast.parse(src)
|
||
main_fn = next(n for n in ast.walk(tree)
|
||
if isinstance(n, ast.FunctionDef) and n.name == "main")
|
||
literals = {n.value for n in ast.walk(main_fn)
|
||
if isinstance(n, ast.Constant) and isinstance(n.value, str)}
|
||
self.assertIn("set_inference_options", literals,
|
||
"set_inference_options 未掛進 main() 的 dispatch")
|
||
|
||
|
||
# ── handle_inference（分支 + 不回歸）─────────────────────────────────
|
||
class TestHandleInferenceBranching(BridgeStateTestCase):
|
||
def setUp(self):
|
||
super().setUp()
|
||
bridge._device_group = object()
|
||
bridge._model_id = 1784536643
|
||
# 讓 handle_inference 跳過影像 decode / SDK send-receive
|
||
self._patches = [
|
||
mock.patch.object(bridge, "HAS_CV2", False),
|
||
mock.patch.object(bridge.kp, "GenericImageInferenceDescriptor",
|
||
lambda **kw: object(), create=True),
|
||
mock.patch.object(bridge.kp, "GenericInputNodeImage",
|
||
lambda **kw: object(), create=True),
|
||
mock.patch.object(bridge.kp, "ImageFormat",
|
||
mock.Mock(KP_IMAGE_FORMAT_RGB565="rgb565"),
|
||
create=True),
|
||
mock.patch.object(bridge.kp.inference, "generic_image_inference_send",
|
||
lambda *a, **k: None, create=True),
|
||
]
|
||
for p in self._patches:
|
||
p.start()
|
||
self.addCleanup(p.stop)
|
||
|
||
def _run(self, arrays, image_b64="Zm9v"):
|
||
result = FakeResult(arrays)
|
||
with mock.patch.object(bridge.kp.inference,
|
||
"generic_image_inference_receive",
|
||
lambda *a, **k: result, create=True), \
|
||
patch_retrieve(result):
|
||
return bridge.handle_inference({"image_base64": image_b64})
|
||
|
||
def test_classification_result_shape(self):
|
||
bridge._model_type = "classification"
|
||
bridge._model_labels = ["剪刀", "石頭", "布"]
|
||
res = self._run([np.array([[0.1, 5.0, 0.2]])])
|
||
self.assertEqual(res["taskType"], bridge.TASK_TYPE_CLASSIFICATION)
|
||
self.assertEqual(res["detections"], [])
|
||
self.assertEqual(res["classifications"][0]["label"], "石頭")
|
||
self.assertEqual(res["classifications"][0]["classIndex"], 1)
|
||
self.assertIn("timestamp", res)
|
||
self.assertIn("latencyMs", res)
|
||
|
||
def test_classification_without_labels_uses_enum(self):
|
||
bridge._model_type = "classification"
|
||
bridge._model_labels = None
|
||
res = self._run([np.array([[0.1, 5.0, 0.2]])])
|
||
self.assertEqual(res["classifications"][0]["label"], "class_1")
|
||
|
||
def test_legacy_resnet18_still_goes_classification(self):
|
||
bridge._model_type = "resnet18"
|
||
res = self._run([np.array([[1.0, 2.0]])])
|
||
self.assertEqual(res["taskType"], bridge.TASK_TYPE_CLASSIFICATION)
|
||
|
||
def test_detection_task_type_is_object_detection_not_detection(self):
|
||
"""R-4：detection 路徑回傳統一值域。"""
|
||
bridge._model_type = "tiny_yolov3"
|
||
bridge._model_labels = None
|
||
res = self._run([np.zeros((1, 255, 7, 7))])
|
||
self.assertEqual(res["taskType"], bridge.TASK_TYPE_OBJECT_DETECTION)
|
||
self.assertEqual(res["classifications"], [])
|
||
|
||
def test_detection_path_unaffected_by_absent_labels(self):
|
||
"""既有 detection 流程不回歸：沒 labels 時仍走 COCO fallback。
|
||
|
||
斷言到實際 label 字串，而非只檢查 key 存在 —— 否則 detection
|
||
完全壞掉回空陣列也會通過（m-2）。
|
||
"""
|
||
bridge._model_type = "tiny_yolov3"
|
||
bridge._model_labels = None
|
||
res = self._run([make_yolo_tensor(16)])
|
||
self.assertEqual([d["label"] for d in res["detections"]], ["dog"])
|
||
|
||
def test_detection_with_truncated_labels_falls_back_to_coco(self):
|
||
"""m-2 / C-1 情境釘死：生產環境 labels 永遠不是 None。
|
||
|
||
models.json 每個 detection model 都帶一份 **只有 10 筆的截斷 COCO**。
|
||
_resolve_label 是「逐項 fallback」而非「整體覆蓋」：index < 10 用注入的
|
||
labels、index >= 10 落到完整 80 類 COCO_CLASSES。
|
||
|
||
曾有審查認為這會讓 index >= 10 退化成 class_N（實際不會）。此測試把
|
||
正確行為釘死：未來若有人把 _resolve_label 改成「有 labels 就整份取代」
|
||
或拿掉 detection parser 的 fallback_labels，這裡會立刻失敗。
|
||
"""
|
||
bridge._model_type = "tiny_yolov3"
|
||
bridge._model_labels = list(TRUNCATED_COCO_LABELS)
|
||
self.assertEqual(len(bridge._model_labels), 10)
|
||
|
||
for class_id, expected in ((16, "dog"), (23, "giraffe"), (56, "chair"),
|
||
(79, "toothbrush")):
|
||
res = self._run([make_yolo_tensor(class_id)])
|
||
self.assertEqual([d["label"] for d in res["detections"]], [expected],
|
||
msg=f"class_id={class_id} should resolve to {expected}")
|
||
|
||
def test_detection_with_truncated_labels_uses_injection_in_range(self):
|
||
"""對照組：index < 10 必須用注入的 labels（不是永遠走 COCO）。"""
|
||
bridge._model_type = "tiny_yolov3"
|
||
custom = list(TRUNCATED_COCO_LABELS)
|
||
custom[2] = "自訂汽車"
|
||
bridge._model_labels = custom
|
||
|
||
res = self._run([make_yolo_tensor(2)])
|
||
self.assertEqual([d["label"] for d in res["detections"]], ["自訂汽車"])
|
||
|
||
def test_classification_parse_failure_returns_error_not_empty(self):
|
||
"""shape 對不上時必須回 error、不可靜默回空結果。"""
|
||
bridge._model_type = "classification"
|
||
res = self._run([np.zeros((85, 7, 7))])
|
||
self.assertIn("error", res)
|
||
self.assertIn("(85, 7, 7)", res["error"])
|
||
self.assertNotIn("classifications", res)
|
||
|
||
def test_no_model_loaded_returns_error(self):
|
||
bridge._model_id = None
|
||
res = bridge.handle_inference({"image_base64": "Zm9v"})
|
||
self.assertEqual(res["error"], "no model loaded")
|
||
|
||
def test_no_image_returns_error(self):
|
||
bridge._model_type = "classification"
|
||
res = bridge.handle_inference({"image_base64": ""})
|
||
self.assertEqual(res["error"], "no image data provided")
|
||
|
||
|
||
# ── detection parser label injection ─────────────────────────────────
|
||
class TestDetectionLabelInjection(BridgeStateTestCase):
|
||
def test_ssd_defaults_to_face(self):
|
||
self.assertEqual(
|
||
bridge._resolve_label(0, labels=None, fallback_labels=["face"]),
|
||
"face")
|
||
|
||
def test_ssd_uses_injected_label(self):
|
||
self.assertEqual(
|
||
bridge._resolve_label(0, labels=["人臉"], fallback_labels=["face"]),
|
||
"人臉")
|
||
|
||
def test_parsers_accept_labels_kwarg(self):
|
||
import inspect
|
||
for fn in (bridge._parse_yolo_output, bridge._parse_fcos_output,
|
||
bridge._parse_ssd_output):
|
||
self.assertIn("labels", inspect.signature(fn).parameters,
|
||
msg=f"{fn.__name__} missing labels kwarg")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|