fix(actpred): 修复 DFC 连通性检查在 5.3 HAR 下被静默跳过的缺陷

DFC 5.3 的 HAR 不含 .edges 文件,原检查从未真正执行(仅打印
WARNING 跳过),输入断连的 HEF 仍可编译通过。改为解析主 hn 图
(layer output 边)做 BFS 可达性验证,任一输入层不可达输出层即
报错阻断;har_input_layer_names 同步改为数字后缀排序并优先取
主图 hn。新增 3 个单测(排序/通过/断连拦截),全量 76 通过。

本轮 224×224 重编译实测:优化后 HAR 5 输入均可达输出层(此前
64 版 HEF 输入不敏感的连通性隐患已消除);但 NPU 复验输出仍不
敏感,pred_token 边缘 scale=1965.92 表明抖动校准未生效,根因
待查(见后续提交)。
This commit is contained in:
qing committed 2026-08-13 05:13:18 +00:00
1 parent a6e1781185
commit 8c58770280
2 files changed
+105 -36

No files matched your search

+61
View File
@@ -1217,5 +1217,66 @@ class TestHailoPredictor(unittest.TestCase):
np.testing.assert_allclose(result, [3, 3, 3], atol=1e-6)
class TestDfcCompile(unittest.TestCase):
"""dfc_compile.py 的 HAR 输入层连通性检查(编译前拦截输入断连)"""
@staticmethod
def _make_har(layers):
import json
import tarfile
data = json.dumps(
{"name": "t", "net_params": {}, "layers": layers}).encode()
path = tempfile.mktemp(suffix=".har")
with tarfile.open(path, "w") as tf:
ti = tarfile.TarInfo("net.hn")
ti.size = len(data)
tf.addfile(ti, io.BytesIO(data))
return path
@staticmethod
def _layers(connect_input2=True):
return {
"n/input_layer1": {"type": "input_layer", "input": [],
"output": ["n/conv1"]},
"n/input_layer2": {"type": "input_layer", "input": [],
"output": ["n/conv2"]},
"n/conv1": {"type": "conv", "input": ["n/input_layer1"],
"output": ["n/out"]},
"n/conv2": {"type": "conv", "input": ["n/input_layer2"],
"output": ["n/out" if connect_input2 else "n/dead"]},
"n/dead": {"type": "conv", "input": ["n/conv2"], "output": []},
"n/out": {"type": "output_layer",
"input": ["n/conv1"] + (["n/conv2"] if connect_input2 else []),
"output": []},
}
def test_input_layer_names_sorted_by_suffix(self):
from dfc_compile import har_input_layer_names
har = self._make_har(self._layers())
try:
names = har_input_layer_names(har)
finally:
os.unlink(har)
self.assertEqual(names, ["n/input_layer1", "n/input_layer2"])
def test_connectivity_pass(self):
from dfc_compile import check_input_connectivity
har = self._make_har(self._layers())
try:
check_input_connectivity(har, ["n/input_layer1", "n/input_layer2"])
finally:
os.unlink(har)
def test_disconnected_input_blocked(self):
from dfc_compile import check_input_connectivity
har = self._make_har(self._layers(connect_input2=False))
try:
with self.assertRaises(SystemExit) as ctx:
check_input_connectivity(har, ["n/input_layer1", "n/input_layer2"])
finally:
os.unlink(har)
self.assertIn("n/input_layer2", str(ctx.exception))
if __name__ == "__main__":
unittest.main(verbosity=2)