import argparse
import glob
import sys
import os
import re
import datetime
import json
from typing import List
import opdesc_parser
import const_var
PYF_PATH = os.path.dirname(os.path.realpath(__file__))
IMPL_HEAD = '''#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
Copyright (c) Huawei Technologies Co., Ltd. {}-{}. All rights reserved.
"""
import os, sys
import ctypes
import json
import shutil
from asc_op_compile_base.common.platform import get_soc_spec
from asc_op_compile_base.common.utils import para_check
from asc_op_compile_base.asc_op_compiler import compile_op, replay_op, check_op_cap, generalize_op_params, get_code_channel, OpInfo
from asc_op_compile_base.common.buildcfg import get_default_build_config
from asc_op_compile_base.common.buildcfg import get_current_build_config
from asc_op_compile_base.common import register as tbe_register
{}PYF_PATH = os.path.dirname(os.path.realpath(__file__))
__version__ = '2.0.0'
DTYPE_MAP = {{"float32": ["DT_FLOAT", "float"],
"float16": ["DT_FLOAT16", "half"],
"int8": ["DT_INT8", "int8_t"],
"int16": ["DT_INT16", "int16_t"],
"int32": ["DT_INT32", "int32_t"],
"int64": ["DT_INT64", "int64_t"],
"uint1": ["DT_UINT1", "uint1b_t"],
"uint8": ["DT_UINT8", "uint8_t"],
"uint16": ["DT_UINT16", "uint16_t"],
"uint32": ["DT_UINT32", "uint32_t"],
"uint64": ["DT_UINT64", "uint64_t"],
"bool": ["DT_BOOL", "bool"],
"double": ["DT_DOUBLE", "double"],
"dual": ["DT_DUAL", "unknown"],
"dual_sub_int8": ["DT_DUAL_SUB_INT8", "unknown"],
"dual_sub_uint8": ["DT_DUAL_SUB_UINT8", "unknown"],
"string": ["DT_STRING", "unknown"],
"complex32": ["DT_COMPLEX32", "complex32"],
"complex64": ["DT_COMPLEX64", "complex64"],
"complex128": ["DT_COMPLEX128", "unknown"],
"qint8": ["DT_QINT8", "unknown"],
"qint16": ["DT_QINT16", "unknown"],
"qint32": ["DT_QINT32", "unknown"],
"quint8": ["DT_QUINT8", "unknown"],
"quint16": ["DT_QUINT16", "unknown"],
"resource": ["DT_RESOURCE", "unknown"],
"string_ref": ["DT_STRING_REF", "unknown"],
"int4": ["DT_INT4", "int4b_t"],
"bfloat16": ["DT_BF16", "bfloat16_t"],
"float8_e5m2": ["DT_FLOAT8_E5M2", "fp8_e5m2_t"],
"float8_e4m3fn": ["DT_FLOAT8_E4M3FN", "fp8_e4m3fn_t"],
"hifloat8":["DT_HIFLOAT8", "hifloat8_t"],
"float8_e8m0":["DT_FLOAT8_E8M0", "fp8_e8m0_t"],
"float4_e2m1":["DT_FLOAT4_E2M1", "fp4x2_e2m1_t"],
"float4_e1m2":["DT_FLOAT4_E1M2", "fp4x2_e1m2_t"],
"hifloat4": ["DT_HIFLOAT4", "hifloat4x2_t"],
"hifloat4_scale": ["DT_HIFLOAT4_SCALE", "hif4_scale"],
"int2": ["DT_INT2", "int2b_t"]}}
def add_dtype_fmt_option_single(x, x_n, is_ref: bool = False):
options = []
x_fmt = x.get("format")
x_dtype = x.get("dtype")
x_n_in_kernel = x_n + '_REF' if is_ref else x_n
options.append("-DDTYPE_{{n}}={{t}}".format(n=x_n_in_kernel, t=DTYPE_MAP.get(x_dtype)[1]))
options.append("-DORIG_DTYPE_{{n}}={{ot}}".format(n=x_n_in_kernel, ot=DTYPE_MAP.get(x_dtype)[0]))
options.append("-DFORMAT_{{n}}=FORMAT_{{f}}".format(n=x_n_in_kernel, f=x_fmt))
return options
def get_dtype_fmt_options(__inputs__, __outputs__):
options = []
input_names = {}
output_names = {}
unique_param_name_set = set()
for idx, x in enumerate(__inputs__):
if x is None:
continue
x_n = input_names[idx].upper()
unique_param_name_set.add(x_n)
options += add_dtype_fmt_option_single(x, x_n)
for idx, x in enumerate(__outputs__):
if x is None:
continue
x_n = output_names[idx].upper()
if x_n in unique_param_name_set:
options += add_dtype_fmt_option_single(x, x_n, True)
else:
options += add_dtype_fmt_option_single(x, x_n)
return options
def load_dso(so_path):
try:
ctypes.CDLL(so_path)
except OSError as error :
print(error)
raise RuntimeError("cannot open %s" %(so_path))
else:
print("load so succ ", so_path)
def get_shortsoc_compile_option(compile_option_list: list, shortsoc:str):
compile_options = []
if shortsoc in compile_option_list:
compile_options.extend(compile_option_list[shortsoc])
if '__ALLSOC__' in compile_option_list:
compile_options.extend(compile_option_list['__ALLSOC__'])
return compile_options
def get_kernel_source(src_file, dir_snake, dir_ex):
src_ex = os.path.join(PYF_PATH, "..", "ascendc", dir_ex, src_file)
if os.path.exists(src_ex):
return src_ex
src = os.environ.get('BUILD_KERNEL_SRC')
if src and os.path.exists(src):
return src
src = os.path.join(PYF_PATH, "..", "ascendc", dir_snake, src_file)
if os.path.exists(src):
return src
src = os.path.join(PYF_PATH, src_file)
if os.path.exists(src):
return src
src = os.path.join(PYF_PATH, "..", "ascendc", dir_snake, dir_snake + ".cpp")
if os.path.exists(src):
return src
src = os.path.join(PYF_PATH, "..", "ascendc", dir_ex, dir_ex + ".cpp")
if os.path.exists(src):
return src
src = os.path.join(PYF_PATH, "..", "ascendc", os.path.splitext(src_file)[0], src_file)
if os.path.exists(src):
return src
return src_ex
'''
IMPL_API = """
@tbe_register.register_operator("{}", trans_bool_to_s8=False)
@para_check.check_op_params({})
def {}({}, kernel_name="{}", impl_mode=""):
{}
if get_current_build_config("enable_op_prebuild"):
return
__inputs__, __outputs__, __attrs__ = _build_args({})
options = get_dtype_fmt_options(__inputs__, __outputs__)
options += ["-x", "cce"]
bisheng = os.environ.get('BISHENG_REAL_PATH')
if bisheng is None:
bisheng = shutil.which("bisheng")
if bisheng is not None:
bisheng_path = os.path.dirname(bisheng)
tikcpp_path = os.path.realpath(os.path.join(bisheng_path, "..", "..", "tikcpp"))
else:
tikcpp_path = os.path.realpath("/usr/local/Ascend/cann/compiler/tikcpp")
options.append("-I" + tikcpp_path)
options.append("-I" + os.path.join(tikcpp_path, "..", "..", "include"))
options.append("-I" + os.path.join(tikcpp_path, "tikcfw"))
options.append("-I" + os.path.join(tikcpp_path, "tikcfw", "impl"))
options.append("-I" + os.path.join(tikcpp_path, "tikcfw", "interface"))
options.append("-I" + os.path.join(tikcpp_path,
"..", "..", "..", "latest", os.uname().machine+"-linux", "asc", "atcos"))
options.append("-I" + os.path.join(PYF_PATH, "..", "ascendc", "common"))
if impl_mode == "high_performance":
options.append("-DHIGH_PERFORMANCE=1")
elif impl_mode == "high_precision":
options.append("-DHIGH_PRECISION=1")
if get_current_build_config("enable_deterministic_mode") == 1:
options.append("-DDETERMINISTIC_MODE=1")
else:
options.append("-DDETERMINISTIC_MODE=0")
custom_compile_options = {},
custom_all_compile_options = {},
soc_version = get_soc_spec("SOC_VERSION")
soc_short = get_soc_spec("SHORT_SOC_VERSION").lower()
custom_compile_options_soc = get_shortsoc_compile_option(custom_compile_options[0], soc_short)
custom_all_compile_options_soc = get_shortsoc_compile_option(custom_all_compile_options[0], soc_short)
options += custom_all_compile_options_soc
options += custom_compile_options_soc
origin_func_name = "{}"
ascendc_src_dir_ex = "{}"
ascendc_src_dir = "{}"
ascendc_src_file = "{}"
src = get_kernel_source(ascendc_src_file, ascendc_src_dir, ascendc_src_dir_ex)
"""
REPLAY_OP_API = """
print("start replay Acend C Operator {}, kernel name is {}")
tikreplay_codegen_path = tikcpp_path + "/tikreplaylib/lib"
tikreplay_stub_path = tikcpp_path + "/tikreplaylib/lib/" + soc_version
print("start load libtikreplaylib_codegen.so and libtikreplaylib_stub.so")
codegen_so_path = tikreplay_codegen_path + "/libtikreplaylib_codegen.so"
replaystub_so_path = tikreplay_stub_path + "/libtikreplaylib_stub.so"
if PYF_PATH.endswith("dynamic"):
op_replay_path = os.path.join(PYF_PATH, "..", "..", "op_replay")
else:
op_replay_path = os.path.join(PYF_PATH, "..", "op_replay")
replayapi_so_path = os.path.join(op_replay_path, "libreplay_{}_" + soc_short + ".so")
load_dso(codegen_so_path)
load_dso(replaystub_so_path)
load_dso(replayapi_so_path)
op_type = "{}"
entry_obj = os.path.join(op_replay_path, "{}_entry_" + soc_short + ".o")
code_channel = get_code_channel(src, kernel_name, op_type, options)
op_info = OpInfo(kernel_name = kernel_name, op_type = op_type, inputs = __inputs__, outputs = __outputs__,\\
attrs = __attrs__, impl_mode = impl_mode, param_type_dynamic = {})
res, msg = replay_op(op_info, entry_obj, code_channel, src, options)
if not res:
print("call replay op failed for %s and get into call compile op" %(msg))
compile_op(src, origin_func_name, op_info, options, code_channel, '{}')
"""
COMPILE_OP_API = """
print("start compile Ascend C operator {}. kernel name is " + kernel_name)
op_type = "{}"
code_channel = get_code_channel(src, kernel_name, op_type, options)
op_info = OpInfo(kernel_name = kernel_name, op_type = op_type, inputs = __inputs__, outputs = __outputs__,\\
attrs = __attrs__, impl_mode = impl_mode, origin_inputs=[{}], origin_outputs = [{}],\\
param_type_dynamic = {}, mc2_ctx = {}, param_type_list = {}, init_value_list = {},\\
output_shape_depend_on_compute = {})
compile_op(src, origin_func_name, op_info, options, code_channel, '{}')
"""
COMPILE_OP_API_BUILT_IN = """
print("start compile Ascend C operator {}. kernel name is " + kernel_name)
op_type = "{}"
code_channel = get_code_channel(src, kernel_name, op_type, options)
op_info = OpInfo(kernel_name = kernel_name, op_type = op_type, inputs = __inputs__, outputs = __outputs__,\\
attrs = __attrs__, impl_mode = impl_mode, origin_inputs=[{}], origin_outputs = [{}],\\
param_type_dynamic = {}, mc2_ctx = {}, param_type_list = {}, init_value_list = {},\\
output_shape_depend_on_compute = {})
op_compile_option = '{}'
dat_path = os.path.realpath(os.path.join(PYF_PATH, "..", "..", "ascendc_impl.dat"))
if os.path.exists(dat_path):
# dat file exists: built in hidden src file online compiling process. append vfs compile option in compile_op
abs_rel_kernel_src_path = "{}"
extend_options = {{}}
extend_options['opp_kernel_hidden_dat_path'] = dat_path
compile_op(abs_rel_kernel_src_path, origin_func_name, op_info, options, code_channel, op_compile_option,\\
extend_options)
else:
# dat file does not exist, run original compile cmd
compile_op(src, origin_func_name, op_info, options, code_channel, op_compile_option)
"""
PYPTO_COMPILE_OP_API = """
print("start compile pypto Operator {}. kernel name is " + kernel_name)
op_type = "{}"
if not os.path.exists(src):
raise RuntimeError("pypto DSL kernel not found: " + src)
code_channel = -1
op_info = OpInfo(kernel_name = kernel_name, op_type = op_type, inputs = __inputs__, outputs = __outputs__,\\
attrs = __attrs__, impl_mode = impl_mode, origin_inputs=[{}], origin_outputs = [{}],\\
param_type_dynamic = {}, mc2_ctx = {}, param_type_list = {}, init_value_list = {},\\
output_shape_depend_on_compute = {})
pypto_compile_op(src, origin_func_name, op_info, options, code_channel, '{}')
"""
PYPTO_IMPORT_HEADER = """
import inspect
from unittest.mock import MagicMock as _PyptoMagicMock
sys.modules["torch"] = _PyptoMagicMock(name="torch")
sys.modules["torch_npu"] = _PyptoMagicMock(name="torch_npu")
from pypto_pro.runtime.opc.pypto_compile import pypto_compile_op
"""
SUP_API = """
def {}({}, impl_mode=""):
__inputs__, __outputs__, __attrs__ = _build_args({})
ret_str = check_op_cap("{}", "{}", __inputs__, __outputs__, __attrs__)
ret_dict = json.loads(ret_str)
err_code = ret_dict.get("ret_code")
sup = "Unknown"
reason = "Unknown reason"
if err_code is not None:
if err_code == 0:
sup = "True"
reason = ""
elif err_code == 1:
sup = "False"
reason = ret_dict.get("reason")
else:
sup = "Unknown"
reason = ret_dict.get("reason")
return sup, reason
"""
CAP_API = """
def {}({}, impl_mode=""):
__inputs__, __outputs__, __attrs__ = _build_args({})
result = check_op_cap("{}", "{}", __inputs__, __outputs__, __attrs__)
return result.decode("utf-8")
"""
GLZ_API = """
@tbe_register.register_param_generalization("{}")
def {}_generalization({}, generalize_config=None):
__inputs__, __outputs__, __attrs__ = _build_args({})
ret_str = generalize_op_params("{}", __inputs__, __outputs__, __attrs__, generalize_config)
return [json.loads(ret_str)]
"""
ATTR_DEFAULT = {
"bool": "False",
"int": "0",
"float": "0.0",
"list_int": "[]",
"list_float": "[]",
"list_bool": "[]",
"list_list_int": "[[]]",
"str": "",
}
def optype_snake(origin_str):
temp_str = origin_str[0].lower() + origin_str[1:]
new_str = re.sub(r"([A-Z])", r"_\1", temp_str).lower()
return new_str
def optype_snake_ex(s):
snake_case = ""
for i, c in enumerate(s):
if i == 0:
snake_case += c.lower()
elif c.isupper():
if s[i - 1] != "_":
if not s[i - 1].isupper():
snake_case += "_"
elif s[i - 1].isupper() and (i + 1) < len(s) and s[i + 1].islower():
snake_case += "_"
snake_case += c.lower()
else:
snake_case += c
return snake_case
class AdpBuilder(opdesc_parser.OpDesc):
def __init__(self: any, op_type: str):
self.argsdefv = []
self.op_compile_option: str = "{}"
self.is_pypto = False
super().__init__(op_type)
def write_adapt(
self: any, impl_path, path: str, op_compile_option_all: list = None
):
self._build_paradefault()
if os.environ.get("BUILD_BUILTIN_OPP") != "1" and impl_path != "":
if not self.kernel_src:
self.kernel_src = self.op_file
kernel_ext = ".py" if self.is_pypto else ".cpp"
if self.op_file.endswith("_apt"):
op_dir = self.op_file.replace("_apt", "")
src_file = os.path.join(impl_path, op_dir, self.kernel_src + kernel_ext)
elif self.op_file.endswith("_910b"):
op_dir = self.op_file.replace("_910b", "")
src_file = os.path.join(impl_path, op_dir, self.kernel_src + kernel_ext)
else:
src_file = os.path.join(
impl_path, self.op_file, self.kernel_src + kernel_ext
)
if not os.path.exists(src_file):
print(
f"[ERROR]: operator: {self.op_file} source file: {src_file} does not found, please check."
)
return
out_path = os.path.abspath(path)
if self.dynamic_shape and not out_path.endswith("dynamic"):
out_path = os.path.join(path, "dynamic")
os.makedirs(out_path, exist_ok=True)
adpfile = os.path.join(out_path, self.op_file + ".py")
self._gen_op_compile_option(op_compile_option_all)
with os.fdopen(os.open(adpfile, const_var.WFLAGS, const_var.WMODES), "w") as fd:
self._write_head(fd)
self._write_argparse(fd)
self._write_impl(fd, impl_path)
if self.op_chk_support:
self._write_cap("check_supported", fd)
self._write_cap("get_op_support_info", fd)
if self.op_fmt_sel:
self._write_cap("op_select_format", fd)
self._write_cap("get_op_specific_info", fd)
if self.op_range_limit == "limited" or self.op_range_limit == "dynamic":
self._write_glz(fd)
def _gen_op_compile_option(self: any, op_compile_option_all: list = None):
if op_compile_option_all is not None:
if self.op_type in op_compile_option_all:
self.op_compile_option = json.dumps(op_compile_option_all[self.op_type])
elif "__all__" in op_compile_option_all:
self.op_compile_option = json.dumps(op_compile_option_all["__all__"])
def _ip_argpack(self: any, default: bool = True) -> list:
args = []
for i in range(len(self.input_name)):
arg = self.input_name[i]
if default and self.argsdefv[i] is not None:
arg += "=" + self.argsdefv[i]
args.append(arg)
return args
def _op_argpack(self: any, default: bool = True) -> list:
args = []
argidx = len(self.input_name)
for i in range(len(self.output_name)):
arg = self.output_name[i]
if default and self.argsdefv[i + argidx] is not None:
arg += "=" + self.argsdefv[i + argidx]
args.append(arg)
return args
def _attr_argpack(self: any, default: bool = True) -> list:
args = []
argidx = len(self.input_name) + len(self.output_name)
for i in range(len(self.attr_list)):
att = self.attr_list[i]
arg = att
if default and self.argsdefv[i + argidx] is not None:
if self.attr_val.get(att).get("type") == "str":
arg += '="' + self.argsdefv[i + argidx] + '"'
elif self.attr_val.get(att).get("type") == "bool":
arg += "=" + self.argsdefv[i + argidx].capitalize()
elif self.attr_val.get(att).get("type") == "list_bool":
arg += (
"="
+ "["
+ ", ".join(
word.strip().capitalize()
for word in self.argsdefv[i + argidx].strip("[]").split(",")
)
+ "]"
)
else:
arg += "=" + self.argsdefv[i + argidx]
args.append(arg)
return args
def _build_paralist(self: any, default: bool = True) -> str:
args = []
args.extend(self._ip_argpack(default))
args.extend(self._op_argpack(default))
args.extend(self._attr_argpack(default))
return ", ".join(args)
def _io_parachk(self: any, types: list, type_name: str) -> list:
chk = []
for iot in types:
if iot == "optional":
ptype = "OPTION"
else:
ptype = iot.upper()
chk.append("para_check.{}_{}".format(ptype, type_name))
return chk
def _attr_parachk(self: any) -> list:
chk = []
for att in self.attr_list:
att_type = self.attr_val.get(att).get("type").upper()
chk.append("para_check.{}_ATTR_{}".format("OPTION", att_type))
return chk
def _build_parachk(self: any) -> str:
chk = []
chk.extend(self._io_parachk(self.input_type, "INPUT"))
chk.extend(self._io_parachk(self.output_type, "OUTPUT"))
chk.extend(self._attr_parachk())
chk.append("para_check.KERNEL_NAME")
return ", ".join(chk)
def _build_virtual(self: any) -> str:
virt_exp = []
for index in range(len(self.input_name)):
if self.input_virt.get(index) is None:
continue
val = []
val.append('"param_name":"{}"'.format(self.input_name[index]))
val.append('"index":{}'.format(index))
val.append('"dtype":"{}"'.format(self.input_dtype[index].split(",")[0]))
val.append('"format":"{}"'.format(self.input_fmt[index].split(",")[0]))
val.append('"ori_format":"{}"'.format(self.input_fmt[index].split(",")[0]))
val.append('"paramType":"optional"')
val.append('"shape":[1]')
val.append('"ori_shape":[1]')
virt_exp.append(
" " + self.input_name[index] + " = {" + ",".join(val) + "}"
)
if len(virt_exp) > 0:
return "\n".join(virt_exp)
else:
return " # do ascendc build step"
def _build_mc2_ctx(self: any):
if len(self.mc2_ctx) != 0:
return '["' + '", "'.join(self.mc2_ctx) + '"]'
return "[]"
def _build_paradefault(self: any):
optional = False
argtypes = []
argtypes.extend(self.input_type)
argtypes.extend(self.output_type)
in_idx = 0
for atype in argtypes:
if atype == "optional":
optional = True
if optional:
self.argsdefv.append("None")
else:
self.argsdefv.append(None)
in_idx += 1
for attr in self.attr_list:
atype = self.attr_val.get(attr).get("paramType")
if atype == "optional":
optional = True
attrval = self.attr_val.get(attr).get("defaultValue")
if attrval is not None:
optional = True
self.argsdefv.append(attrval)
continue
if optional:
self.argsdefv.append(
ATTR_DEFAULT.get(self.attr_val.get(attr).get("type"))
)
else:
self.argsdefv.append(None)
def _write_head(self: any, fd: object):
now = datetime.datetime.now()
curr_year = now.year
former_year = curr_year - 1
import_header = PYPTO_IMPORT_HEADER if self.is_pypto else ""
fd.write(
IMPL_HEAD.format(
former_year,
curr_year,
import_header,
self.input_ori_name,
self.output_ori_name,
)
)
def _write_argparse(self: any, fd: object):
args = self._build_paralist(False)
fd.write("def _build_args({}):\n".format(args))
fd.write(" __inputs__ = []\n")
fd.write(" for arg in [{}]:\n".format(", ".join(self.input_name)))
fd.write(" if arg != None:\n")
fd.write(" if isinstance(arg, (list, tuple)):\n")
fd.write(" if len(arg) == 0:\n")
fd.write(" continue\n")
fd.write(" __inputs__.append(arg[0])\n")
fd.write(" else:\n")
fd.write(" __inputs__.append(arg)\n")
fd.write(" else:\n")
fd.write(" __inputs__.append(arg)\n")
fd.write(" __outputs__ = []\n")
fd.write(" for arg in [{}]:\n".format(", ".join(self.output_name)))
fd.write(" if arg != None:\n")
fd.write(" if isinstance(arg, (list, tuple)):\n")
fd.write(" if len(arg) == 0:\n")
fd.write(" continue\n")
fd.write(" __outputs__.append(arg[0])\n")
fd.write(" else:\n")
fd.write(" __outputs__.append(arg)\n")
fd.write(" else:\n")
fd.write(" __outputs__.append(arg)\n")
fd.write(" __attrs__ = []\n")
for attr in self.attr_list:
fd.write(" if {} != None:\n".format(attr))
fd.write(" attr = {}\n")
fd.write(' attr["name"] = "{}"\n'.format(attr))
fd.write(
' attr["dtype"] = "{}"\n'.format(
self.attr_val.get(attr).get("type")
)
)
fd.write(' attr["value"] = {}\n'.format(attr))
fd.write(" __attrs__.append(attr)\n")
fd.write(" return __inputs__, __outputs__, __attrs__\n")
def _get_kernel_source(self: any, kernel_src_dir, src_file, dir_snake, dir_ex):
src_ex = os.path.join(kernel_src_dir, dir_ex, src_file)
if os.path.exists(src_ex):
return src_ex
src = os.environ.get("BUILD_KERNEL_SRC")
if src and os.path.exists(src):
return src
src = os.path.join(kernel_src_dir, dir_snake, src_file)
if os.path.exists(src):
return src
src = os.path.join(kernel_src_dir, src_file)
if os.path.exists(src):
return src
src = os.path.join(kernel_src_dir, dir_snake, dir_snake + ".cpp")
if os.path.exists(src):
return src
src = os.path.join(kernel_src_dir, dir_ex, dir_ex + ".cpp")
if os.path.exists(src):
return src
src = os.path.join(kernel_src_dir, os.path.splitext(src_file)[0], src_file)
if os.path.exists(src):
return src
return src_ex
def _write_impl(self: any, fd: object, impl_path: str = ""):
argsdef = self._build_paralist()
argsval = self._build_paralist(False)
pchk = self._build_parachk()
if len(self.kern_name) > 0:
kern_name = self.kern_name
else:
kern_name = self.op_intf
if not self.kernel_src:
self.kernel_src = self.op_file
src = self.kernel_src + (".py" if self.is_pypto else ".cpp")
virt_exprs = self._build_virtual()
fd.write(
IMPL_API.format(
self.op_type,
pchk,
self.op_intf,
argsdef,
kern_name,
virt_exprs,
argsval,
self.custom_compile_options,
self.custom_all_compile_options,
self.op_intf,
optype_snake_ex(self.op_type),
optype_snake(self.op_type),
src,
)
)
if self.op_replay_flag:
fd.write(
REPLAY_OP_API.format(
self.op_type,
kern_name,
self.op_file,
self.op_type,
self.op_file,
self.param_type_dynamic,
self.op_compile_option,
)
)
elif self.is_pypto:
fd.write(
PYPTO_COMPILE_OP_API.format(
self.op_type,
self.op_type,
", ".join(self.input_name),
", ".join(self.output_name),
self.param_type_dynamic,
self._build_mc2_ctx(),
self.input_type + self.output_type,
self.output_init_value,
self.output_shape_depend_on_compute,
self.op_compile_option,
)
)
else:
if os.environ.get("BUILD_BUILTIN_OPP") == "1":
relative_kernel_src_path = os.path.realpath(
self._get_kernel_source(
impl_path,
src,
optype_snake(self.op_type),
optype_snake_ex(self.op_type),
)
)
abs_rel_kernel_src_path = os.path.join(
"/", os.path.relpath(relative_kernel_src_path, impl_path)
)
fd.write(
COMPILE_OP_API_BUILT_IN.format(
self.op_type,
self.op_type,
", ".join(self.input_name),
", ".join(self.output_name),
self.param_type_dynamic,
self._build_mc2_ctx(),
self.input_type + self.output_type,
self.output_init_value,
self.output_shape_depend_on_compute,
self.op_compile_option,
abs_rel_kernel_src_path,
)
)
else:
fd.write(
COMPILE_OP_API.format(
self.op_type,
self.op_type,
", ".join(self.input_name),
", ".join(self.output_name),
self.param_type_dynamic,
self._build_mc2_ctx(),
self.input_type + self.output_type,
self.output_init_value,
self.output_shape_depend_on_compute,
self.op_compile_option,
)
)
def _write_cap(self: any, cap_name: str, fd: object):
argsdef = self._build_paralist()
argsval = self._build_paralist(False)
if cap_name == "check_supported":
fd.write(SUP_API.format(cap_name, argsdef, argsval, cap_name, self.op_type))
else:
fd.write(CAP_API.format(cap_name, argsdef, argsval, cap_name, self.op_type))
def _write_glz(self: any, fd: object):
argsdef = self._build_paralist()
argsval = self._build_paralist(False)
fd.write(
GLZ_API.format(self.op_type, self.op_intf, argsdef, argsval, self.op_type)
)
def _parse_pypto_ops(value: str) -> set:
"""Split a comma/semicolon separated op_file list into a set, dropping empties."""
return {o for o in (value or "").replace(";", ",").split(",") if o}
def write_scripts(
cfgfile: str,
cfgs: dict,
dirs: dict,
ops: list = None,
op_compile_option: list = None,
pypto_ops: set = None,
):
batch_lists = cfgs.get(const_var.REPLAY_BATCH).split(";")
iterator_lists = cfgs.get(const_var.REPLAY_ITERATE).split(";")
pypto_ops = pypto_ops or _parse_pypto_ops(os.environ.get("PYPTO_OPS"))
file_map = {}
op_descs = opdesc_parser.get_op_desc(
cfgfile,
batch_lists,
iterator_lists,
AdpBuilder,
ops,
dirs.get(const_var.AUTO_GEN_DIR),
)
for op_desc in op_descs:
op_desc.is_pypto = op_desc.op_file in pypto_ops
op_desc.write_adapt(
dirs.get(const_var.CFG_IMPL_DIR),
dirs.get(const_var.CFG_OUT_DIR),
op_compile_option,
)
file_map[op_desc.op_type] = op_desc.op_file
return file_map
class OpFileNotExistsError(Exception):
"""File does not exist error."""
def __str__(self) -> str:
return (
f"File aic-*-ops-info.ini does not exist in directory {super().__str__()}"
)
def get_ops_info_files(opsinfo_dir: List[str]) -> List[str]:
"""Get all ops info files."""
ops_info_files = []
for _dir in opsinfo_dir:
ops_info_files.extend(glob.glob(f"{_dir}/aic-*-ops-info.ini"))
return sorted(ops_info_files)
def parse_args(argv):
"""Command line parameter parsing"""
parser = argparse.ArgumentParser()
parser.add_argument("argv", nargs="+")
parser.add_argument("--opsinfo-dir", nargs="*", default=None)
parser.add_argument("--pypto-ops", nargs="?", const="", default="")
return parser.parse_args(argv)
if __name__ == "__main__":
args = parse_args(sys.argv)
if len(args.argv) <= 6:
raise RuntimeError("arguments must greater equal than 6")
rep_cfg = {}
rep_cfg[const_var.REPLAY_BATCH] = args.argv[2]
rep_cfg[const_var.REPLAY_ITERATE] = args.argv[3]
cfg_dir = {}
cfg_dir[const_var.CFG_IMPL_DIR] = args.argv[4]
cfg_dir[const_var.CFG_OUT_DIR] = args.argv[5]
cfg_dir[const_var.AUTO_GEN_DIR] = args.argv[6]
ops_infos = []
if args.opsinfo_dir:
ops_infos.extend(get_ops_info_files(args.opsinfo_dir))
if not ops_infos:
raise OpFileNotExistsError(args.opsinfo_dir)
else:
ops_infos.append(args.argv[1])
pypto_ops = _parse_pypto_ops(args.pypto_ops)
for ops_info in ops_infos:
write_scripts(cfgfile=ops_info, cfgs=rep_cfg, dirs=cfg_dir, pypto_ops=pypto_ops)
