diff --git a/actpred/dfc_compile.py b/actpred/dfc_compile.py index e3b2252..59ad912 100644 --- a/actpred/dfc_compile.py +++ b/actpred/dfc_compile.py @@ -25,6 +25,7 @@ import argparse import json import os +import re import shutil import subprocess import sys @@ -43,56 +44,63 @@ def hailo(*args, venv_bin: str): return subprocess.run(cmd, env=env, check=True) -def har_input_layer_names(har_path: str) -> list: - """从 HAR（tar 包）中读 hn 文件，按 input_layerN 排序返回层名""" +def _har_hn_layers(har_path: str) -> dict: + """解包 HAR 并读取主 hn 文件的 layers（优化产物含 fp/native 变体，取主图）""" with tempfile.TemporaryDirectory() as tmp: with tarfile.open(har_path) as tf: tf.extractall(tmp) hn_files = [f for f in os.listdir(tmp) if f.endswith(".hn")] - with open(os.path.join(tmp, hn_files[0])) as f: - hn = json.load(f) - layers = [(name, l) for name, l in hn["layers"].items() - if l["type"] == "input_layer"] - layers.sort(key=lambda x: x[0]) - return [name for name, _ in layers] + if not hn_files: + raise ValueError(f"{har_path} 中无 .hn 文件") + main = [f for f in hn_files + if not f.endswith(".fp.hn") and not f.endswith(".native.hn")] + hn_file = main[0] if main else hn_files[0] + with open(os.path.join(tmp, hn_file)) as f: + return json.load(f)["layers"] + + +def _input_layer_index(name: str) -> int: + m = re.search(r"input_layer(\d+)$", name) + return int(m.group(1)) if m else 0 + + +def har_input_layer_names(har_path: str) -> list: + """从 HAR 主 hn 图读输入层名，按 input_layerN 数字后缀排序""" + layers = _har_hn_layers(har_path) + names = [n for n, l in layers.items() if l["type"] == "input_layer"] + names.sort(key=_input_layer_index) + return names def check_input_connectivity(har_path: str, input_layers: list): - """检查 HAR 中输入层是否都有下游边 + """检查 HAR 图中每个输入层是否可达任一输出层 输入层被优化断开时编译照常成功，但 HEF 输出恒定值且无任何报错 - （2026-08-13 实测踩坑），必须在编译前拦截。.edges 文件缺失或格式 - 无法识别时仅告警不阻断。 + （2026-08-13 实测踩坑），必须在编译前拦截。DFC 5.3 的 HAR 不含 + .edges 文件，连通性改从主 hn 图的 layer output 边 BFS 判定。 """ - with tempfile.TemporaryDirectory() as tmp: - with tarfile.open(har_path) as tf: - tf.extractall(tmp) - edge_files = [f for f in os.listdir(tmp) if f.endswith(".edges")] - if not edge_files: - print(f"WARNING: {har_path} 无 .edges 文件，跳过输入连通性检查") - return - with open(os.path.join(tmp, edge_files[0])) as f: - edges = json.load(f) - - if not isinstance(edges, list): - print(f"WARNING: .edges 不是列表（{type(edges).__name__}），" - f"跳过连通性检查") + layers = _har_hn_layers(har_path) + adj = {n: l.get("output", []) for n, l in layers.items()} + targets = {n for n, l in layers.items() if l["type"] == "output_layer"} + if not targets: + print(f"WARNING: {har_path} 图中无输出层，跳过连通性检查") return - adjacency = set() - for e in edges: - if isinstance(e, dict) and "from" in e: - adjacency.add(e["from"]) - elif isinstance(e, (list, tuple)) and len(e) == 2: - adjacency.add(e[0]) - else: - print(f"WARNING: 无法识别 .edges 条目格式: {e!r}，跳过连通性检查") - return - disconnected = [n for n in input_layers if n not in adjacency] + disconnected = [] + for name in input_layers: + seen, stack = set(), [name] + while stack: + cur = stack.pop() + if cur in seen: + continue + seen.add(cur) + stack.extend(adj.get(cur, [])) + if not (seen & targets): + disconnected.append(name) if disconnected: sys.exit( - f"错误: {har_path} 中输入层无下游边（输入被优化断开，" + f"错误: {har_path} 中输入层不可达输出层（输入被优化断开，" f"编译出的 HEF 将输出恒定值）: {disconnected}") - print(f"输入连通性检查通过: {len(input_layers)} 个输入层均有下游边") + print(f"输入连通性检查通过: {len(input_layers)} 个输入层均可达输出层") def main(): diff --git a/actpred/test_actpred.py b/actpred/test_actpred.py index d6fb537..00e2664 100644 --- a/actpred/test_actpred.py +++ b/actpred/test_actpred.py @@ -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)