# -*- coding: utf-8 -*- """ Created on Sun Nov 10 18:13:57 2019 @author: sun Fortran 调用Python时的模块，包含一些必要的函数和类。如果需要改动增加新的函数与类，则需要写好之后重新 编译DLL，如果只是更改函数内部的处理方式，则不需要。 这个版本的程序传递过来的数据是最原始的流场变量数据，特征的构建和顺序需要在BuildFeatures函数里面完成 模型的预测赋值也直接在pred函数里面完成，所以一定不要改pred函数里面x的值，否则数据可能会传递不过去 这里只是不分区涡粘模型的写法，其他的类似 如果需要提高运行速度，一个优化策略是有选择性的取消属性访问，比如把np.vstack改为form numpy import vstack，而且要尽量使用局部变量 ----2020.1.25s 调用华为om模型执行推理 2022.4.20 """ import numpy as np import time import acl import constants as const import acllite_utils as utils from acllite_logger import log_error, log_info, log_warning from acllite_resource import resource_list, AclLiteResource from omaclitemodel import AclLiteModel from constant import ACL_MEM_MALLOC_HUGE_FIRST, \ ACL_MEMCPY_HOST_TO_DEVICE, ACL_MEMCPY_DEVICE_TO_HOST, \ NPY_FLOAT32, ACL_SUCCESS, NPY_FLOAT16,ACL_PROF_ACL_API, \ ACL_PROF_TASK_TIME, ACL_PROF_AICPU_TRACE, ACL_PROF_AICORE_METRICS, ACL_PROF_L2CACHE buffer_method = { "in": acl.mdl.get_input_size_by_index, "out": acl.mdl.get_output_size_by_index } # create the dictionary mapping ctypes to np dtypes ctype2dtype = {} # Intertypes for prefix in ('int', 'uint'): for log_bytes in range(4): ctype = '%s%d_t' % (prefix, 8 * (2 ** log_bytes)) dtype = '%s%d' % (prefix[0], 2 ** log_bytes) ctype2dtype[ctype] = np.dtype(dtype) # Floating point types ctype2dtype['float'] = np.dtype('f4') ctype2dtype['double'] = np.dtype('f8') ctype2dtype['int'] = np.dtype('int32') def check_ret(message, ret): if ret != ACL_SUCCESS: raise Exception("{} failed ret={}" .format(message, ret)) def GetDataFromPointer(ffi, pointer): """ 将指针的数据获取出来 """ T = ffi.getctype(ffi.typeof(pointer).item) value = np.frombuffer(ffi.buffer(pointer, ffi.sizeof(T)), ctype2dtype[T]) value = value.item() return value class BFNet(object): def __init__(self, device_id, model_path, model_path_2, model_path_3): self.device_id = device_id # int self.model_path = model_path # string self.model_path_2 = model_path_2 self.model_path_3 = model_path_3 self.model_id = None # pointer self.model_id_2 = None self.model_id_3 = None self.context = None # pointer self.input_data = [] self.output_data = [] self.output_data_2 = [] self.output_data_3 = [] self.model_desc = None # pointer when using self.model_desc_2 = None self.model_desc_3 = None self.load_input_dataset = None self.load_output_dataset = None self.load_output_dataset_2 = None self.load_output_dataset_3 = None self.prof_config = None self.init_resource() def __del__(self): print("Releasing resources stage:") ret = acl.mdl.unload(self.model_id) # check_ret("acl.mdl.unload", ret) ret = acl.mdl.unload(self.model_id_2) # ret = acl.mdl.unload(self.model_id_3) if self.model_desc: acl.mdl.destroy_desc(self.model_desc) self.model_desc = None if self.model_desc_2: acl.mdl.destroy_desc(self.model_desc_2) self.model_desc_2 = None if self.model_desc_3: acl.mdl.destroy_desc(self.model_desc_3) self.model_desc_3 = None while self.input_data: item = self.input_data.pop() ret = acl.rt.free(item["buffer"]) # check_ret("acl.rt.free", ret) while self.output_data: item = self.output_data.pop() ret = acl.rt.free(item["buffer"]) # check_ret("acl.rt.free", ret) while self.output_data_2: item = self.output_data_2.pop() ret = acl.rt.free(item["buffer"]) # check_ret("acl.rt.free", ret) while self.output_data_3: item = self.output_data_3.pop() ret = acl.rt.free(item["buffer"]) # check_ret("acl.rt.free", ret) if self.context: ret = acl.rt.destroy_context(self.context) # check_ret("acl.rt.destroy_context", ret) self.context = None ret = acl.rt.reset_device(self.device_id) # check_ret("acl.rt.reset_device", ret) ret = acl.finalize() # check_ret("acl.finalize", ret) # print('Resources released successfully.') def init_resource(self): # print("init resource stage:") ret = acl.init() # check_ret("acl.init", ret) ret = acl.rt.set_device(self.device_id) # check_ret("acl.rt.set_device", ret) self.context, ret = acl.rt.create_context(self.device_id) # check_ret("acl.rt.create_context", ret) # load_model self.model_id, ret = acl.mdl.load_from_file(self.model_path) # check_ret("acl.mdl.load_from_file", ret) # print("model_id:{}".format(self.model_id)) self.model_id_2, ret = acl.mdl.load_from_file(self.model_path_2) # check_ret("acl.mdl.load_from_file", ret) # print("model_id_2:{}".format(self.model_id_2)) self.model_id_3, ret = acl.mdl.load_from_file(self.model_path_3) check_ret("acl.mdl.load_from_file", ret) # print("model_id_2:{}".format(self.model_id_2)) self.model_desc = acl.mdl.create_desc() self._get_model_info() self.model_desc_2 = acl.mdl.create_desc() self._get_model_info_2() self.model_desc_3 = acl.mdl.create_desc() self._get_model_info_3() # print("init resource success") def _get_model_info(self,): ret = acl.mdl.get_desc(self.model_desc, self.model_id) # check_ret("acl.mdl.get_desc", ret) input_size = acl.mdl.get_num_inputs(self.model_desc) # print("input_size") # print(input_size) output_size = acl.mdl.get_num_outputs(self.model_desc) # print("output_size") # print(output_size) self._gen_data_buffer(input_size, des="in") self._gen_data_buffer(output_size, des="out") def _get_model_info_2(self,): ret = acl.mdl.get_desc(self.model_desc_2, self.model_id_2) # check_ret("acl.mdl.get_desc", ret) input_size = acl.mdl.get_num_inputs(self.model_desc_2) # print("model_2_input_size") # print(input_size) output_size = acl.mdl.get_num_outputs(self.model_desc_2) # print("model2_output_size") # print(output_size) # self._gen_data_buffer(input_size, des="in") self._gen_data_buffer_2(output_size, des="out") def _get_model_info_3(self,): ret = acl.mdl.get_desc(self.model_desc_3, self.model_id_3) # check_ret("acl.mdl.get_desc", ret) input_size = acl.mdl.get_num_inputs(self.model_desc_3) # print("model_3_input_size") # print(input_size) output_size = acl.mdl.get_num_outputs(self.model_desc_3) # print("model_3_output_size") # print(output_size) # self._gen_data_buffer(input_size, des="in") self._gen_data_buffer_3(output_size, des="out") def _gen_data_buffer(self, size, des): func = buffer_method[des] for i in range(size): # check temp_buffer dtype temp_buffer_size = func(self.model_desc, i) temp_buffer, ret = acl.rt.malloc(temp_buffer_size, ACL_MEM_MALLOC_HUGE_FIRST) # check_ret("acl.rt.malloc", ret) if des == "in": self.input_data.append({"buffer": temp_buffer, "size": temp_buffer_size}) elif des == "out": self.output_data.append({"buffer": temp_buffer, "size": temp_buffer_size}) def _gen_data_buffer_2(self, size, des): func = buffer_method[des] for i in range(size): # check temp_buffer dtype temp_buffer_size = func(self.model_desc_2, i) temp_buffer, ret = acl.rt.malloc(temp_buffer_size, ACL_MEM_MALLOC_HUGE_FIRST) # check_ret("acl.rt.malloc", ret) # # if des == "in": # self.input_data.append({"buffer": temp_buffer, # "size": temp_buffer_size}) # elif des == "out": self.output_data_2.append({"buffer": temp_buffer, "size": temp_buffer_size}) def _gen_data_buffer_3(self, size, des): func = buffer_method[des] for i in range(size): # check temp_buffer dtype temp_buffer_size = func(self.model_desc_3, i) temp_buffer, ret = acl.rt.malloc(temp_buffer_size, ACL_MEM_MALLOC_HUGE_FIRST) # check_ret("acl.rt.malloc", ret) # # if des == "in": # self.input_data.append({"buffer": temp_buffer, # "size": temp_buffer_size}) # elif des == "out": self.output_data_3.append({"buffer": temp_buffer, "size": temp_buffer_size}) def _data_interaction(self, dataset, policy=ACL_MEMCPY_HOST_TO_DEVICE): temp_data_buffer = self.input_data \ if policy == ACL_MEMCPY_HOST_TO_DEVICE \ else self.output_data_3 if len(dataset) == 0 and policy == ACL_MEMCPY_DEVICE_TO_HOST: for item in self.output_data_3: begin = time.time() temp, ret = acl.rt.malloc_host(item["size"]) end = time.time() # print(f"ACL_MEMCPY_HOST_TO_DEVICE malloc = {end-begin}") # print("item[size]") # print(item["size"]) if ret != 0: raise Exception("can't malloc_host ret={}".format(ret)) dataset.append({"size": item["size"], "buffer": temp}) for i, item in enumerate(temp_data_buffer): if policy == ACL_MEMCPY_HOST_TO_DEVICE: # print(f"i is {i}") a = item["size"] # print(f"size is {a}") # print(i) # print(dataset.shape) ptr,_ = acl.util.numpy_contiguous_to_ptr(dataset[i]) begin = time.time() ret = acl.rt.memcpy(item["buffer"], item["size"], ptr, item["size"], policy) end = time.time() # print(f"ACL_MEMCPY_HOST_TO_DEVICE = {end-begin}") check_ret("acl.rt.memcpy", ret) else: # print(f"i is {i}") a = item["size"] # print(f"size is {a}") ptr = dataset[i]["buffer"] begin = time.time() ret = acl.rt.memcpy(ptr, item["size"], item["buffer"], item["size"], policy) end = time.time() # print(f"ACL_MEMCPY_DEVICE_TO_HOST = {end-begin}") check_ret("acl.rt.memcpy", ret) def _gen_dataset(self, type_str="in"): dataset = acl.mdl.create_dataset() temp_dataset = None if type_str == "in": self.load_input_dataset = dataset temp_dataset = self.input_data elif type_str == "out": self.load_output_dataset = dataset temp_dataset = self.output_data elif type_str == "out2": self.load_output_dataset_2 = dataset temp_dataset = self.output_data_2 elif type_str == "out3": self.load_output_dataset_3 = dataset temp_dataset = self.output_data_3 for item in temp_dataset: data = acl.create_data_buffer(item["buffer"], item["size"]) _, ret = acl.mdl.add_dataset_buffer(dataset, data) if ret != ACL_SUCCESS: ret = acl.destroy_data_buffer(data) # check_ret("acl.destroy_data_buffer", ret) def _data_from_host_to_device(self, images): # print("data interaction from host to device") # copy images to device begin = time.time() # print("images is ", images) # print(len(images)) self._data_interaction(images, ACL_MEMCPY_HOST_TO_DEVICE) end = time.time() # print(f"_data_from_host_to_device = {end-begin}") # load input data into model self._gen_dataset("in") # load output data into model self._gen_dataset("out") # load second output data into model self._gen_dataset("out2") # # load third output data into model self._gen_dataset("out3") # # print("data interaction from host to device success") def _data_from_device_to_host(self): # print("data interaction from device to host") res = [] # copy device to host begin = time.time() self._data_interaction(res, ACL_MEMCPY_DEVICE_TO_HOST) end = time.time() # print(f"_data_from_device_to_host = {end-begin}") # print("data interaction from device to host success") # result = self.get_result(res,self.model_desc_3, NPY_FLOAT32) result = self.get_result(res, self.model_desc_3, NPY_FLOAT32) # self._print_result(result) # free host memory # for item in res: # ptr = item['buffer'] # ret = acl.rt.free_host(ptr) # check_ret('acl.rt.free_host', ret) return result def run(self, images): start = time.time() self._data_from_host_to_device(images) end = time.time() # print(f"_data_from_host_to_device = {end-start}") # self.start_profile() self.forward() end2 = time.time() # print(f"forward = {end2-end}") # self.end_profile(self.prof_config) result = self._data_from_device_to_host() end3 = time.time() # print(f"_data_from_device_to_host = {end3-end2}") return result def forward(self): # print('execute stage:') start = time.time() ret = acl.mdl.execute(self.model_id, self.load_input_dataset, self.load_output_dataset) end1 = time.time() # print(f"first om time={end1 - start}") # check_ret("acl.mdl.execute", ret) # start = time.time() ret = acl.mdl.execute(self.model_id_2, self.load_output_dataset, self.load_output_dataset_2) end2 = time.time() # print(f"second om time={end2 - end1}") # check_ret("acl.mdl.execute", ret) # start = time.time() ret = acl.mdl.execute(self.model_id_3, self.load_output_dataset_2, self.load_output_dataset_3) end3 = time.time() # print(f"third om time={end3 - end2}") # check_ret("acl.mdl.execute", ret) self._destroy_databuffer() # print('execute stage success') def _print_result(self, result): print(result) def _destroy_databuffer(self): # for dataset in [self.load_input_dataset, self.load_output_dataset, self.load_output_dataset_2, self.load_output_dataset_3]: for dataset in [self.load_input_dataset, self.load_output_dataset, self.load_output_dataset_2, self.load_output_dataset_3]: # for dataset in [self.load_input_dataset, self.load_output_dataset, self.load_output_dataset_2]: if not dataset: continue number = acl.mdl.get_dataset_num_buffers(dataset) for i in range(number): data_buf = acl.mdl.get_dataset_buffer(dataset, i) if data_buf: ret = acl.destroy_data_buffer(data_buf) # check_ret("acl.destroy_data_buffer", ret) ret = acl.mdl.destroy_dataset(dataset) # check_ret("acl.mdl.destroy_dataset", ret) def get_result(self, output_data, model_desc, data_type): result = [] dims, ret = acl.mdl.get_cur_output_dims(model_desc, 0) # check_ret("acl.mdl.get_cur_output_dims", ret) out_dim = dims['dims'] # print(tuple(out_dim)) for temp in output_data: ptr = temp["buffer"] # print(temp) # 转化为float32类型的数据 data = acl.util.ptr_to_numpy(ptr, tuple(out_dim), data_type) # print(data) result.append(data) return result def asarray(ffi, x_ptr, shape1_ptr, shape2_ptr): """ x_ptr : Fortran传递过来的数组的地址 size_ptr : Fortran数组的大小的地址 因为传递过来的都是地址，因此需要先把地址中的数值取出来，才可以进行数据处理 """ shape = [GetDataFromPointer(ffi, pointer) for pointer in (shape1_ptr, shape2_ptr)] length = np.prod(shape) # Get the canonical C type of the elments of ptr as a string T = ffi.getctype(ffi.typeof(x_ptr).item) x = np.frombuffer(ffi.buffer(x_ptr, length * ffi.sizeof(T)), ctype2dtype[T]).reshape(shape, order='C') return x def read_data(file_path): with open(file_path, 'r') as f: return np.array([float(x) for x in f.read().split()]).reshape(-1, 1) class pythonmodel(object): def __init__(self): """ 类初始化 """ self.sca_min = np.loadtxt('./model/sca_min.dat').reshape(-1, 1) self.sca_max = np.loadtxt('./model/sca_max.dat').reshape(-1, 1) self.temp = np.ones((1300000, 17)) self.model1_path = "./model/bf_op.om" self.model2_path = "./model/model16.om" self.model3_path = "./model/post_op.om" self.flag = 0 print('model loaded!') def modelInit(self, zone_id): """ om模型初始化 """ self.dev_id = zone_id % 8 # self.DSTURB_min = dis_min self.bfnet = BFNet(self.dev_id, self.model1_path, self.model2_path, self.model3_path) def BuildFeatures(self, x): """ 基于原始变量，构建模型需要的输入特征 x : 包含原始变量，依次为：Ru, U, V, W, P, Ux, Uy, Uz, Vx, Vy, Vz, Wx, Wy, Wz, Px, Py, Pz, x, y, z, dis, ma, aoa, re """ ma = x[:, -2] self.Re = x[:, -1] print(self.Re.shape) # eps = 1e-13 # aoa = x[:, -2] Ru, P = x[:, 0], x[:, 3] * ma ** 2 U, V = x[:, 1], x[:, 2] Ux, Uy, Uz = x[:, 4], x[:, 5], x[:, 6] Vx, Vy, Vz = x[:, 7], x[:, 8], x[:, 9] Wx, Wy, Wz = x[:, 10], x[:, 11], x[:, 12] # Px, Py, Pz = x[:, 14] * ma ** 2, x[:, 15] * ma ** 2, x[:, 16] * ma ** 2 Y = x[:, 13] dis = x[:, 14] self.dis = dis # q1 = np.sqrt(U**2 + V**2 + W**2) q1 = U # / np.sqrt(U**2 + V**2 + W**2) # q1_2 = V / np.sqrt(U**2 + V**2 + W**2) w1 = 0.5 * (Wy - Vz) w2 = 0.5 * (Uz - Wx) w3 = 0.5 * (Vx - Uy) RF = np.sqrt(w1 ** 2 + w2 ** 2 + w3 ** 2) # q2 = np.log(np.abs(RF) + 1.0) q3 = (P / Ru ** 1.4) - 1.0 label = np.where(dis < 0.001, 1, 2) sigY = np.sign(Y) q5 = np.arctan(sigY * V / U) S11 = Ux S12 = 0.5 * (Uy + Vx) S13 = 0.5 * (Uz + Wx) S22 = Vy S23 = 0.5 * (Vz + Wy) S33 = Wz SF = np.sqrt(S11 ** 2 + S12 ** 2 + S13 ** 2 + S22 ** 2 + S23 ** 2 + S33 ** 2) # q7 = np.log(np.abs(SF) + 1.0) q8 = dis ** 2 * RF * (1 - np.tanh(dis)) q9 = dis ** 2 * SF * (1 - np.tanh(dis)) Dref0 = 1.0 / np.sqrt(self.Re) Dref1 = np.min([dis, Dref0], axis=0) Dref2 = np.max([dis, Dref0], axis=0) expfunc = 2.71828 ** (np.sqrt(Dref1 / (dis))) q10 = expfunc * np.sqrt(Dref0 / (Dref2)) q11 = (RF ** 2 - SF ** 2) / (RF ** 2 + SF ** 2) data = np.vstack((q1, label, RF, q3, q5, SF, q8, q9, q10, q11)).T # 输入特征顺序 col = data.shape[1] x = (data - self.sca_min[0:col, 0]) / (self.sca_max[0:col, 0] - self.sca_min[0:col, 0]) return x def GetFueatures(self, x): ma = x[:, -3] self.Re = x[:, -1] aoa = x[:, -2] Ru, P = x[:, 0], x[:, 3] U, V = x[:, 1], x[:, 2] Ux, Uy = x[:, 4], x[:, 5] Vx, Vy = x[:, 6], x[:, 7] Y, self.dis = x[:, 8], x[:, 9] data = [ma, aoa, self.Re, self.dis, Ux, U, V, Uy, Vx, Vy, Y, P, Ru] return data def ACT(self, x, low, up, length): """ 调用om模型预测涡粘 low: 归一化下界 up: 归一化上界 x: 输入特征，np.array格式 """ x = x.astype(np.float32) ptr, data_out = acl.util.numpy_contiguous_to_ptr(x) pred = self.bfnet.run([data_out])[0] pred = pred[0:length] return pred def Pred(self, x): """ 模型预测涡粘 """ start = time.time() length = x.shape[0] self.temp[0:length, :] = x[:, :] low, up = 0.0, 1.0 pred = self.ACT(self.temp, low, up, length) x[:, 0] = pred[:].copy() end = time.time() def post_op(pred, low, high) : sca_min = np.loadtxt('./model/sca_min.dat').reshape(-1, ) sca_max = np.loadtxt('./model/sca_max.dat').reshape(-1, ) Re = 11710000 pred = np.clip(pred[:, 0], low, high).reshape(-1, 1) mut_min = 8.539709649871904494e-07 mut_max = 8.296507258753202052e+01 pred = pred * (mut_max - mut_min) + mut_min pred = pred * (Re / 1e6) return pred if __name__ == "__main__" : model_2 = pythonmodel() model_2.modelInit(5) np.random.seed(4) x = np.random.rand(1300000, 17).astype(np.float32) y = x.copy() model_2.Pred(y) pred_2 = y[:, 0] print("**********") print(pred_2) print(type(pred_2)) print(pred_2.mean()) print(pred_2.std()) print(pred_2.shape)