"""Auto-generate 03_参数定义.yaml from CANN header prototype + ops-* md doc."""
import json, os, re, sys
import yaml
BASE = "/home/30039325/Vector_fuzz_cpp_testing"
INC = "/usr/local/Ascend/ascend-toolkit/latest/x86_64-linux/include/aclnnop"
INDEX = "/tmp/opencode/op_index.json"
DTYPE_MAP = {"FLOAT32": "FLOAT", "FLOAT": "FLOAT", "FLOAT16": "FLOAT16", "BFLOAT16": "BFLOAT16",
"DOUBLE": "DOUBLE", "INT8": "INT8", "INT16": "INT16", "INT32": "INT32", "INT64": "INT64",
"UINT8": "UINT8", "UINT16": "UINT16", "UINT32": "UINT32", "UINT64": "UINT64", "BOOL": "BOOL",
"COMPLEX64": "COMPLEX64", "COMPLEX128": "COMPLEX128"}
SPECIAL_API = {
"aclnnSparseFlashAttention_A5": {
"header": "aclnn_sparse_flash_attention.h",
"get_workspace_size": "aclnnSparseFlashAttentionGetWorkspaceSize",
"func_api": "aclnnSparseFlashAttention",
},
}
def parse_prototype(header_path, func_name):
content = open(header_path, encoding="utf-8", errors="ignore").read()
m = re.search(re.escape(func_name) + r"\s*\((.*?)\)", content, re.DOTALL)
if not m:
return None
parts = [p.strip() for p in m.group(1).split(",") if p.strip()]
params = []
for p in parts:
pm = re.match(r"^(const\s+)?([\w:]+)\s*(\*{0,2})\s*(\w+)$", p)
if not pm:
continue
const, typ, stars, name = pm.group(1), pm.group(2), pm.group(3), pm.group(4)
if name in ("workspaceSize", "executor", "opExecutor", "workspace", "stream") or typ == "aclOpExecutor":
continue
params.append({"const": bool(const), "type": typ, "ptr": bool(stars), "stars": stars, "name": name})
return params
def parse_md_info(md_path, api, func_api):
"""Parse dtype/dims/io/format from md parameter table for this api."""
try:
content = open(md_path, encoding="utf-8", errors="ignore").read()
except Exception:
return {}
sec = None
for name in (func_api, api):
sec = re.search(r"##\s*" + re.escape(name) + r"GetWorkspaceSize(.*?)(?=\n## |\Z)", content, re.DOTALL)
if sec:
break
if not sec:
sec = re.search(r"GetWorkspaceSize(.*?)(?=\n## |\Z)", content, re.DOTALL)
if not sec:
return {}
text = sec.group(1)
info = {}
for row in re.findall(r"<tr[^>]*>(.*?)</tr>", text, re.DOTALL):
cells = re.findall(r"<td[^>]*>(.*?)</td>", row, re.DOTALL)
if len(cells) < 5:
continue
strip = lambda s: re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", s)).strip()
name_cell = strip(cells[0])
name = re.split(r"[((]", name_cell)[0].strip()
if not name or name == "参数名":
continue
io = strip(cells[1]) if len(cells) > 1 else ""
desc_text = strip(cells[2]) if len(cells) > 2 else ""
usage = strip(cells[3]) if len(cells) > 3 else ""
dtype_cell = strip(cells[4]) if len(cells) > 4 else ""
fmt = strip(cells[5]) if len(cells) > 5 else ""
dims = strip(cells[6]) if len(cells) > 6 else ""
dtypes = []
for tok in re.split(r"[、,,/\s]+", dtype_cell):
tok = tok.strip().upper()
if tok in DTYPE_MAP and DTYPE_MAP[tok] not in dtypes:
dtypes.append(DTYPE_MAP[tok])
info[name] = {"io": io, "dtypes": dtypes, "format": fmt, "dims": dims,
"usage": usage, "desc": desc_text}
return info
def extract_array_count(text):
"""Extract fixed array length from md text (数组长度恒为N/长度为N/size为N...)."""
if not text:
return None
for pat in (r"数组长度恒为(\d+)", r"数组长度必须为(\d+)", r"数组长度为(\d+)",
r"长度恒为(\d+)", r"长度必须为(\d+)", r"长度只能是(\d+)", r"长度为(\d+)",
r"size为(\d+)", r"Size为(\d+)"):
m = re.search(pat, text)
if m:
n = int(m.group(1))
if 1 <= n <= 8:
return n
return None
def extract_enum_values(text):
"""Extract enum values from md text (描述+使用说明列).
String enums: 取值可以是"static"或"dynamic" -> ['static', 'dynamic']
Int enums: 取值可以是0、1、2 / 取值为0或1 -> [0, 1, 2]
Returns (is_enum, values) or (False, None).
"""
if not text or "取值" not in text:
return False, None
strs = re.findall(r"\"([\w./-]+)\"", text)
if len(set(strs)) >= 2:
return True, sorted(set(strs))
m = re.search(r"取值(?:可以是|为|:|:)\s*(\d+(?:\s*[、,,/或]\s*\d+)+)", text)
if m:
ints = [int(x) for x in re.findall(r"\d+", m.group(1))]
if len(set(ints)) >= 2:
return True, sorted(set(ints))
return False, None
def parse_dims(dims_str):
if not dims_str:
return None
s = dims_str.replace("-", "-").strip()
m = re.match(r"^(\d+)\s*-\s*(\d+)$", s)
if m:
lo, hi = int(m.group(1)), int(m.group(2))
if lo <= hi <= 8:
return [lo, hi]
if "8" in s and ("不超过" in s or "维" in s):
return [0, 8]
return None
def value_default(cpp, name):
n = name.lower()
if cpp in ("double", "float"):
return "0.000001" if "eps" in n else "0.5"
if cpp == "bool":
return "false"
if cpp == "const char*":
return '"dynamic"' if "mode" in n else '""'
if cpp == "aclDataType":
return "ACL_FLOAT"
if cpp == "aclFormat":
return "ACL_FORMAT_ND"
if "axis" in n or n == "dim" or "kdim" in n:
return "-1"
return "1"
def gen_yaml(api):
idx = json.load(open(INDEX))
ent = idx.get(api, {})
sp = SPECIAL_API.get(api, {})
header = sp.get("header") or ent.get("header")
md = ent.get("md")
func_api = sp.get("func_api") or api
ws_name = sp.get("get_workspace_size") or (func_api + "GetWorkspaceSize")
if not header:
return None, "no header found"
header_path = os.path.join(INC, header)
if not os.path.isfile(header_path):
return None, "header file missing: " + header
params = parse_prototype(header_path, ws_name)
if params is None:
return None, "prototype not found in header"
if not any(p["type"] in ("aclTensor", "aclTensorList") for p in params):
return None, "no aclTensor/tensorList param"
md_info = parse_md_info(md, api, func_api) if md else {}
out_params = []
for p in params:
name, typ, const = p["name"], p["type"], p["const"]
mi = md_info.get(name, {})
usage_text = " ".join(x for x in (mi.get("usage", ""), mi.get("desc", "")) if x)
if typ == "aclTensor":
io = "output" if not const else ("output" if mi.get("io") == "输出" else "input")
dtypes = mi.get("dtypes") or ["FLOAT", "FLOAT16", "BFLOAT16"]
dims = parse_dims(mi.get("dims", "")) or [0, 8]
entry = {
"name": name, "type": "aclTensor", "is_pointer": True,
"required": "Optional" not in name, "io_type": io,
"format": mi.get("format") or "ND", "dimensions": dims,
"dtype_with_ranges": [{"dtype": d} for d in dtypes],
}
elif typ in ("aclScalar", "aclIntArray", "aclBoolArray", "aclFloatArray"):
entry = {
"name": name, "type": typ, "is_pointer": True,
"required": "Optional" not in name,
"io_type": "input" if const else "output",
"dtype": (mi.get("dtypes") or ["INT64"])[0] if typ == "aclIntArray" else "FLOAT",
}
if typ in ("aclIntArray", "aclBoolArray", "aclFloatArray"):
cnt = extract_array_count(usage_text)
if cnt:
entry["count"] = cnt
entry["value_desc"] = usage_text[:120] or "array param"
if typ == "aclScalar":
is_enum, vals = extract_enum_values(usage_text)
if is_enum:
entry["is_enum"] = True
entry["enum_values"] = vals
elif typ == "aclTensorList":
entry = {
"name": name, "type": "aclTensorList", "is_pointer": True,
"required": "Optional" not in name,
"io_type": "input" if const else "output",
}
elif not p["ptr"] and typ not in ("char",):
cpp = typ
entry = {
"name": name, "type": "value", "cpp_type": cpp,
"default_value": value_default(cpp, name),
"is_pointer": False, "required": True, "io_type": "input",
}
if cpp in ("int64_t", "int32_t", "int8_t", "uint8_t", "uint64_t", "size_t"):
is_enum, vals = extract_enum_values(usage_text)
if is_enum:
entry["is_enum"] = True
entry["enum_values"] = vals
elif typ in ("double", "float", "int64_t", "int32_t", "int8_t", "uint8_t", "uint64_t", "bool", "size_t"):
cpp = typ
entry = {
"name": name, "type": "value", "cpp_type": cpp,
"default_value": value_default(cpp, name),
"is_pointer": False, "required": True, "io_type": "input",
}
elif typ == "char":
entry = {
"name": name, "type": "value", "cpp_type": "const char*",
"default_value": value_default("const char*", name),
"is_pointer": True, "required": "Optional" not in name, "io_type": "input",
}
is_enum, vals = extract_enum_values(usage_text)
if is_enum:
entry["is_enum"] = True
entry["enum_values"] = vals
entry["default_value"] = f'"{vals[0]}"'
else:
entry = {
"name": name, "type": typ, "is_pointer": True,
"required": "Optional" not in name,
"io_type": "input" if const else "output",
}
out_params.append(entry)
doc = {
"api_name": api,
"header": header,
"get_workspace_size": ws_name,
"parameters": out_params,
"error_codes": [
{"code": 161001, "name": "ACLNN_ERR_PARAM_NULLPTR", "description": "必选输入/输出或必选属性是空指针"},
{"code": 161002, "name": "ACLNN_ERR_PARAM_INVALID", "description": "参数dtype/shape/format不在支持范围之内"},
],
}
return doc, None
def main():
api = sys.argv[1]
doc, err = gen_yaml(api)
if err:
print(f"[YAML_FAIL] {api}: {err}")
sys.exit(1)
out_dir = os.path.join(BASE, "output", api, "test-design")
os.makedirs(out_dir, exist_ok=True)
out_path = os.path.join(out_dir, "03_参数定义.yaml")
with open(out_path, "w", encoding="utf-8") as f:
yaml.safe_dump(doc, f, allow_unicode=True, sort_keys=False, default_flow_style=False)
print(f"[YAML_OK] {api}: {len(doc['parameters'])} params -> {out_path}")
if __name__ == "__main__":
main()