AAlice_zhouupdate
3a138af5Add on Sep 1History
#!/usr/bin/env python3
"""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",):
            # any by-value C type: double/int/int64_t/bool/aclDataType/aclFormat...
            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":
            # char* 参数: 指针类, 必须覆盖 nullptr + 枚举非法值场景
            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()