#!/usr/bin/env python3 # coding=utf-8 # Copyright (c) HiSilicon (Shanghai) Technologies Co., Ltd. 2021-2022. All rights reserved. """ * Description: NV binary create. * Create: 2021-11-30 """ import os import re import sys import stat import json import hashlib import struct import shutil import zlib import binascii g_root = os.path.realpath(__file__) g_root = os.path.dirname(g_root) g_root = os.path.realpath(os.path.join(g_root, "..", "..", "..")) sys.path.append(os.path.join(g_root, 'build')) sys.path.append(os.path.join(g_root, 'build', 'script')) # print(g_root) from conf_parser import BuildConfParser, ParserError from build_utils import fn_str_to_int from generate_data_stream import generate_data_stream from ctypes import c_char, c_byte, c_ubyte, c_short, c_ushort, c_int, c_uint, Structure, sizeof TD_CHAR = c_char TD_S8 = c_byte TD_U8 = c_ubyte TD_S16 = c_short TD_U16 = c_ushort TD_S32 = c_int TD_U32 = c_uint g_u8_max = 0xFF g_u16_max = 0xFFFF class KeyHead(Structure): _fields_ = [ ("magic", TD_U8), # Magic number to indicate the start of the item ("valid", TD_U8), # flag to indicate whether the value is valid ("length", TD_U16), # Length of the key_data field in bytes ("type", TD_U8), # Normal (0xFF) or permanent (0x00) or keep (0x04 ) ("upgrade", TD_U8), ("key_id", TD_U16), # The Key ID ("enc_key", TD_U16), # Allows some customisation of the data AES key used, 0x0 - key_data is plaintext, Others - key_data is encrypted ("version", TD_U16), # Version of the key ("rnd", TD_U32) # Key header crc calculated from length ] class NvPageHead(Structure): _fields_ = [ ("id", TD_U16), ("reserved", TD_U8), ("num_pages", TD_U8), ("inverted_details_word", TD_U32), ("last_write", TD_U32), # last_write ("unused", TD_U32), # We want this header to be 4 words long - this allows us to alias anything after it ] # 1.根据指定的alias，合并所有配置文件 # 1.1 解析alias中所有type=Nv的配置。 # 2.如果需要，从合并的配置文件中提取NV项ID，生成ID枚举头文件提供给源码使用（构建过程不允许修改源码，因此该头文件只有需要的时候调用脚本生成） # 3.预编译所有nv结构 # 4.解析配置文件中各个nv项的数据内容，与结构结合，生成bin class BuildNv: def __init__(self, alias, root=None, targets=None, backup=False, use_crc16=False): self.alias = alias self.root = root if root is not None else g_root self.targets = targets self.is_backup = backup self.use_crc16 = use_crc16 self.tmp_path = os.path.join(self.root, json_conf["BUILD_TEMP_PATH"]) self.nv_relative_path = os.path.join(self.root, json_conf["NV_RELATIVE_PATH"]) self.nv_root = os.path.join(self.root, json_conf["NV_DEFAULT_CFG_DIR"]) self.nv_output_dir = os.path.join(self.root, json_conf["OUT_BIN_DIR"]) if not backup: self.nv_output_name = json_conf["OUT_BIN_NAME"] else: self.nv_output_name = json_conf["OUT_BACKUP_BIN_NAME"] self.nv_ver_src_dict = dict() self.nv_ver_dict = dict() self.nv_flash_cfg = None self.nv_cores_ver_bin = dict() self.nv_chip_ver_bin = dict() self.nv_flash_page_index = dict() def set_nv_output_dir(self, path): self.nv_output_dir = path def start_work(self): self._merge_cfgs() self._load_nv_flash_cfg() self._parse_etypes() self._gen_binary() self._create_header() def _merge_cfgs(self): ''' Merge config sources in self.nv_ver_src_dict. This will build the self.nv_ver_dict like the following tree: |--- ver1 : { | "merged_cfg" : json file after merge all nv configuration with the same product type. | "prod_type" : "XXXX" } | chip|---target1---|--- ver2 : | | | |--- ver3 : | | | |--- core : "Each target corresponds to one core." |---targetN... ''' for target in self.alias: if self._nv_ver_prepare(target) is False: continue # print('nv_ver_src_dict: ', self.nv_ver_src_dict) for chip in self.nv_ver_src_dict: src_chip_dict = self.nv_ver_src_dict[chip] # print("src_chip_dict =",src_chip_dict) self.nv_ver_dict[chip] = {} chip_dict = self.nv_ver_dict[chip] for target in src_chip_dict: if chip_dict.get(target) is None: chip_dict[target] = {'core':src_chip_dict[target]['core']} nv_tmp_dir = os.path.join(self.nv_relative_path) for ver_name in src_chip_dict[target]: if ver_name == 'core': continue cfg_file_prefix = os.path.join(nv_tmp_dir, 'cfg', '%s_nv' % (target)) # 生成中间文件路径 # print("cfg_file_prefix = ", cfg_file_prefix) chip_dict[target][ver_name] = self._merge_ver_cfg(cfg_file_prefix, src_chip_dict[target][ver_name]) def _parse_etypes(self): for chip in self.nv_ver_dict: chip_dict = self.nv_ver_dict[chip] for target in chip_dict: # TODO: scons in chip dir or nv_config dir? etypes path depends on scons path nv_tmp_dir = os.path.join(self.tmp_path, target) etypes_path = os.path.join(nv_tmp_dir, "%s.etypes" % target) # 中间文件xxx.etypes路径 if os.path.exists(etypes_path) is not True: # 判断中间文件是否存在，如果不在说明该模块没有被编译，需要加入到编译链接中 etypes_path = os.path.join(self.tmp_path, "etypes", "%s.etypes" % target) if os.path.exists(etypes_path) is not True: msg = "[error] [%s] need add nv_config module in alias! %s" % (target, etypes_path) raise ParserError(msg) stream_gen = generate_data_stream() stream_gen.phase_etypes(etypes_path) chip_dict[target]["stream_gen"] = stream_gen dtabase_txt = os.path.join(self.root, json_conf['DATABASE_TXT_FILE']) shutil.copy(etypes_path, dtabase_txt) def _gen_binary(self): ''' |--- ver1 : binary file of ver1. | (version name : product_type + version name) chip|---core1---|--- ver2 : | | | |--- ver3 : |---coreN... ''' self._gen_binary_prepare() self._gen_target_version_binary() self._gen_chip_nv_binary() def _gen_target_version_binary(self): for chip in self.nv_cores_ver_bin: cores = self.nv_cores_ver_bin[chip] # print("cores =", cores) for core in cores: cores[core] = self._gen_version_binary(self.nv_ver_dict[chip], chip, core) def _gen_version_binary(self, chip_ver_dict, chip, core): ver_binary_dict = dict() # print("chip_ver_dict = ", chip_ver_dict) # 字典信息包含核名字，配置文件路径 if chip_ver_dict is None: return ver_binary_dict for target in chip_ver_dict: if chip_ver_dict[target].get('core') != core: # 判断字典里面的target读出来的core是否和core相同 print("chip_ver_dict[target].get('core') = ", chip_ver_dict[target].get('core')) continue stream_gen = chip_ver_dict[target].get('stream_gen') for ver in chip_ver_dict[target]: # print("ver =", ver) if ver == 'core' or ver == 'stream_gen': continue stream_gen = chip_ver_dict[target].get('stream_gen') cfg_file = chip_ver_dict[target][ver]["merged_cfg"] # print("cfg_file =",cfg_file) stream = self._gen_nv_stream(cfg_file, stream_gen, chip, core) nv_ver_bin = \ os.path.join(self.nv_relative_path, 'bin', '%s_nv.bin' % (core)) # print("nv_ver_bin = ", nv_ver_bin) # 生成nvbin文件的路径 prod_type = chip_ver_dict[target][ver]["prod_type"] prod_type = "all" if prod_type is None else prod_type ver_binary_dict["%s_%s" % (chip, prod_type)] = self._write_binary_to_file(nv_ver_bin, stream) return ver_binary_dict def _write_binary_to_file(self, file_path, stream): if os.path.exists(os.path.dirname(file_path)) is False: os.makedirs(os.path.dirname(file_path)) if os.path.exists(file_path) is True: os.remove(file_path) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL modes = stat.S_IWUSR | stat.S_IRUSR with os.fdopen(os.open(file_path, flags, modes), 'wb') as fout: fout.write(stream) return file_path def _gen_binary_prepare(self): for chip in self.nv_ver_dict: self.nv_chip_ver_bin[chip] = dict() self.nv_cores_ver_bin[chip] = dict() self.nv_flash_page_index[chip] = dict() nv_flash_chip_cfg = self.nv_flash_cfg[chip] cores = nv_flash_chip_cfg["cores"].keys() chip_nv_ver = self.nv_ver_dict[chip] for target in chip_nv_ver: core = chip_nv_ver[target].get('core') for core in cores: if core not in cores: msg = "[error] [%s] [%s] not a core cfg in nv_storage_cfg.json!" % (target, core) raise ParserError(msg) if self.nv_cores_ver_bin[chip].get(core) is None: self.nv_cores_ver_bin[chip][core] = dict() page_size = fn_str_to_int(nv_flash_chip_cfg["size"]["page_size"]) total_size = 0 for core in cores: core_page_nums = nv_flash_chip_cfg['cores'][core]['page_nums'] total_size += core_page_nums * page_size self.nv_flash_page_index[chip][core] = [(sizeof(NvPageHead) + num * page_size, (num + 1) * page_size) \ for num in range(0, core_page_nums)] if total_size > fn_str_to_int(nv_flash_chip_cfg["size"]['flash_size']): msg = "[error] cores size: %s, over total flash size: %s!" % \ (total_size, nv_flash_chip_cfg["size"]['flash_size']) raise ParserError(msg) def _gen_nv_stream(self, cfg_file, stream_gen, chip, core): core_nv_bin = self._init_nv_page_head(chip, core) cfg_data = BuildConfParser(cfg_file).get_conf_data() key_id_list = [] last_key_item_start_addr = 0 for module in cfg_data: for key_item in cfg_data[module]: if key_item == 'module_id': continue key_struct_name = cfg_data[module][key_item].get("structure_type") key_value = cfg_data[module][key_item].get("value") key_attr = cfg_data[module][key_item].get("attributions") key_id = cfg_data[module][key_item].get("key_id") key_id = fn_str_to_int(key_id) if type(key_id) is not int else key_id if key_struct_name is None or key_value is None or key_attr is None or key_value == []: msg = "[error] 'structure_type' 'value' 'attributions' must be configured!" raise ParserError(msg) if key_id in key_id_list: msg = "[error] key id:%d repeated, please check!" % key_id raise ParserError(msg) key_id_list.append(key_id) key_data, key_data_len = stream_gen.generate(key_struct_name, key_value) page_index, key_item_start_addr = self._find_usable_addr(chip, core, key_data, key_data_len) core_nv_bin, key_item_start_addr = \ self._init_key_head(core_nv_bin, key_item_start_addr, key_data_len, key_id, key_attr) core_nv_bin, key_item_start_addr = \ self._set_key_data(core_nv_bin, key_item_start_addr, key_data, key_data_len) core_nv_bin, key_item_start_addr = \ self._set_key_hash(core_nv_bin, key_item_start_addr, key_data_len) self._update_core_index(chip, core, page_index, key_item_start_addr) last_key_item_start_addr = max(last_key_item_start_addr, key_item_start_addr) core_nv_bin = self._set_unused_page(chip, core_nv_bin, last_key_item_start_addr) self._reset_core_index(chip, core) return core_nv_bin def _set_unused_page(self, chip, core_nv_bin, key_item_start_addr): page_size = fn_str_to_int(self.nv_flash_cfg[chip]['size']['page_size']) core_nv_end_addr = (key_item_start_addr + (page_size - 1)) & ~(page_size - 1) for i in range(key_item_start_addr, core_nv_end_addr): core_nv_bin[i] = 0xFF return core_nv_bin[0 : core_nv_end_addr] def _gen_chip_nv_binary(self): for chip in self.nv_cores_ver_bin: chip_bins = self.nv_cores_ver_bin[chip] ver_list = [] for core in chip_bins: ver_list.extend(chip_bins[core].keys()) ver_list = set(ver_list) for ver in ver_list: self.nv_chip_ver_bin[chip][ver] = self._assemble_ver_bins(chip, ver) def _assemble_ver_bins(self, chip, ver): flash_bin = bytearray(fn_str_to_int(self.nv_flash_cfg[chip]['size']['flash_size'])) for i in range(0, len(flash_bin)): flash_bin[i] = 0xFF start_addr = 0 chip_bins = self.nv_cores_ver_bin[chip] cores = self.nv_cores_ver_bin[chip] for core in chip_bins: ver_bin = chip_bins[core].get(ver) flash_bin, start_addr = self._append_file_to_stream(flash_bin, start_addr, chip, core, ver_bin) chip_ver_bin_file = os.path.join(self.nv_output_dir, self.nv_output_name) # TODO: 目前只考虑单核场景去除 NV bin 末尾的无用 FF return self._write_binary_to_file(chip_ver_bin_file, flash_bin[0 : start_addr]) def _append_file_to_stream(self, flash_bin, start_addr, chip, core, ver_bin): core_bin_size = fn_str_to_int(self.nv_flash_cfg[chip]['size']["page_size"]) * \ self.nv_flash_cfg[chip]["cores"][core]['page_nums'] core_nv_bin = b'' if ver_bin is None: core_nv_bin = self._init_nv_page_head(chip, core) else: with open(ver_bin, 'rb') as f: core_nv_bin = f.read() #print('core_nv_bin = ', core_nv_bin) tail_addr = start_addr + len(core_nv_bin) flash_bin[start_addr : tail_addr] = core_nv_bin return flash_bin, tail_addr def _reset_core_index(self, chip, core): nv_flash_chip_cfg = self.nv_flash_cfg[chip] page_size = fn_str_to_int(nv_flash_chip_cfg['size']["page_size"]) core_page_nums = nv_flash_chip_cfg['cores'][core]['page_nums'] self.nv_flash_page_index[chip][core] = [(sizeof(NvPageHead) + num * page_size, (num + 1) * page_size) \ for num in range(0, core_page_nums)] def _update_core_index(self, chip, core, index, addr): (start_addr, page_max_addr) = self.nv_flash_page_index[chip][core][index] if start_addr >= addr or addr > page_max_addr: msg = "[error] addr %s invalid!" % addr raise ParserError(msg) self.nv_flash_page_index[chip][core][index] = (addr, page_max_addr) #print("update page index: \n", self.nv_flash_page_index) def _find_usable_addr(self, chip, core, key_data, key_data_len): page_size = fn_str_to_int(self.nv_flash_cfg[chip]['size'].get('page_size')) key_item_total_len = self._get_key_item_len(key_data_len) if key_item_total_len > page_size - sizeof(NvPageHead): msg = "[error] key over page size !" % key_id raise ParserError(msg) index = 0 for (start_addr, page_max_addr) in self.nv_flash_page_index[chip][core]: if start_addr + key_item_total_len > page_max_addr: index += 1 continue return index, start_addr msg = "[error] no more enough space for [%s]!" % core raise ParserError(msg) def _get_key_type_from_attr(self, key_id, attr): if (attr & 1) and (not (attr & ~1)): return 0xFF elif (attr & 2) and (not (attr & ~2)): return 0x00 elif (attr & 4) and (not (attr & ~4)): return 0xFF else: msg = "[error] attribution config err: [id-%s] [attr-%s] !" % (key_id, attr) raise ParserError(msg) def _get_key_upgrade_from_attr(self, key_id, attr): if (attr & 1) and (not (attr & ~1)): return 0xFF elif (attr & 2) and (not (attr & ~2)): return 0xFF elif (attr & 4) and (not (attr & ~4)): return 0x00 else: msg = "[error] attribution config err: [id-%s] [attr-%s] !" % (key_id, attr) raise ParserError(msg) def _get_key_item_len(self, key_data_len): if key_data_len % 4 != 0: key_data_len += 4 - key_data_len % 4 return sizeof(KeyHead) + 4 + key_data_len def _set_key_hash(self, nv_bin, key_item_start_addr, key_data_len): if key_data_len % 4 != 0: key_data_len += 4 - key_data_len % 4 hash_start_addr = key_item_start_addr - key_data_len - sizeof(KeyHead) hash_end_addr = hash_start_addr + key_data_len + sizeof(KeyHead) if not self.use_crc16: crc32num = zlib.crc32(nv_bin[hash_start_addr : hash_end_addr]) else: crc32num = binascii.crc_hqx(nv_bin[hash_start_addr : hash_end_addr], 0) crc32ret = '{:0>8X}'.format(crc32num) crc32ret = re.sub(r"(?<=\w)(?=(?:\w\w)+$)", " 0x", crc32ret) crc32ret = '0x' + crc32ret crc32list = [int(x,16) for x in crc32ret.split(" ")] sha256bytearray = bytes(crc32list) tail_addr = key_item_start_addr + len(sha256bytearray) nv_bin[key_item_start_addr : tail_addr] = sha256bytearray return nv_bin, tail_addr def _set_key_data(self, nv_bin, key_item_start_addr, key_data, key_data_len): if key_data_len % 4 != 0: for i in range(0, 4 - key_data_len % 4) : key_data += b'\x00' key_data_len += 4 - key_data_len % 4 tail_addr = key_item_start_addr + key_data_len nv_bin[key_item_start_addr : tail_addr] = key_data return nv_bin, tail_addr def _init_key_head(self, nv_bin, key_item_start_addr, key_data_len, key_id, key_attr): nv_key_st = KeyHead.from_buffer(nv_bin[key_item_start_addr:]) nv_key_st.magic = 0xA9 nv_key_st.length = key_data_len nv_key_st.type = self._get_key_type_from_attr(key_id, key_attr) nv_key_st.upgrade = self._get_key_upgrade_from_attr(key_id, key_attr) nv_key_st.key_id = key_id nv_key_st.version = 65535 nv_key_st.enc_key = 0 # TODO: 目前不支持加密，且nv加密部分不由nv脚本做，可能放到初始化？ tail_addr = key_item_start_addr + sizeof(KeyHead) nv_bin[key_item_start_addr : tail_addr] = nv_key_st return nv_bin, tail_addr def _init_nv_page_head(self, chip, core): nv_flash_chip_cfg = self.nv_flash_cfg[chip] default_page_nums = nv_flash_chip_cfg.get('default_page_nums') page_size = fn_str_to_int(nv_flash_chip_cfg['size']['page_size']) page_nums = nv_flash_chip_cfg['cores'][core]['page_nums'] if not self.is_backup: page_id_start = nv_flash_chip_cfg['cores'][core]['page_id_start'] else: page_id_start = '0x34B2' core_nv_size = page_nums * page_size core_nv_bin = bytearray(core_nv_size) for i in range(0, core_nv_size): core_nv_bin[i] = 0xFF for i in range(0, page_nums): start_addr = i * page_size nv_page_head = NvPageHead.from_buffer(core_nv_bin[start_addr:]) nv_page_head.id = fn_str_to_int(page_id_start) nv_page_head.reserved = 1 nv_page_head.num_pages = i nv_page_head.inverted_details_word = ~int.from_bytes(struct.pack('HBB', \ nv_page_head.id, nv_page_head.reserved, nv_page_head.num_pages), 'little') nv_page_head.last_write = 0 nv_page_head.unused = ~nv_page_head.last_write core_nv_bin[start_addr : start_addr + sizeof(NvPageHead)] = nv_page_head return core_nv_bin def _load_nv_flash_cfg(self): self.nv_flash_cfg = dict() for chip in self.nv_ver_dict: cfg_file = os.path.join(self.root, json_conf["NV_TARGET_JSON_PATH"]) self.nv_flash_cfg[chip] = BuildConfParser(cfg_file).get_conf_data() def _add_nv_ver(self, chip, target, core, ver, common_cfg, ver_cfg, prod_type=None): ''' Add version config into self.nv_ver_src_dict. There are three configuration scenarios.One target may correspond to multiple NV versions. |--- ver1: { srcs:[default, common, cfg1], prod_type: } | chip|---target1---|--- ver2: { srcs:[default, cfg2], prod_type: } | | | |--- ver3: { srcs:[default, common], prod_type: } | | | |--- core : "Each target corresponds to one core." |---targetN... ''' ver_cfgs = [] if os.path.exists(common_cfg) is True: ver_cfgs.append(common_cfg) if ver_cfg is not None and os.path.exists(ver_cfg): ver_cfgs.append(ver_cfg) if self.nv_ver_src_dict.get(chip) is None: self.nv_ver_src_dict[chip] = dict() chip_dict = self.nv_ver_src_dict[chip] if chip_dict.get(target) is not None and chip_dict[target].get(ver) is not None: msg = "[error] Ver config Repeate!" raise ParserError(msg) if chip_dict.get(target) is None: chip_dict[target] = {ver:{"srcs":ver_cfgs, 'prod_type': prod_type}} else: chip_dict[target].update({ver:{"srcs":ver_cfgs, 'prod_type': prod_type}}) if chip_dict[target].get('core') is None: chip_dict[target]['core'] = core elif chip_dict[target].get('core') != core: msg = "[error] [%s] core not match!" % target raise ParserError(msg) def _nv_ver_prepare(self, target): ''' 1. Check nv configurations. 2. Add all correct config into self.nv_ver_src_dict.. ''' if type(self.alias[target]) is list: return False target_type = self.alias[target].get("TYPE") if target_type is None or target_type != 'nv': return False core = self.alias[target].get("CORE") if core is None: msg = "[error] core name not exist!" raise ParserError(msg) chip = self.alias[target].get("CHIP") #default_cfg = os.path.join(self.nv_root, '%s_default.json' % core) ''' if chip is None or os.path.exists(default_cfg) is False: msg = "[error] chip name OR %s not exist!" % default_cfg raise ParserError(msg) ''' kernel_name = self.alias[target].get("KERNEL_BIN") if kernel_name is None: msg = "[error] KERNEL is null!" raise ParserError(msg) if self.targets is not None and kernel_name not in self.targets: return False cfgs = self.alias[target].get("COMPONENT") if cfgs is None: msg = "[error] COMPONENT is null!" raise ParserError(msg) prod_type = self.alias[target].get("PRODUCT_TYPE") if prod_type == "": prod_type = None if prod_type is not None and type(prod_type) is not str: msg = "[error] PRODUCT_TYPE must be a string type, one kernel only suuport one product type!" raise ParserError(msg) cfg_dir = os.path.join(self.nv_root, kernel_name) common_cfg = os.path.join(cfg_dir, 'common.json') for cfg in cfgs: cfg_file = os.path.join(cfg_dir, '%s.json' % cfg) if cfg != 'common' else None self._add_nv_ver(chip, kernel_name, core, cfg, common_cfg, cfg_file, prod_type) def _prod_type_filter(self, srcs, prod_type = None): combination = dict() module_dict = dict() for src in srcs: # print("[INFO] nv config src file: ", src) src_conf = BuildConfParser(src).get_conf_data() for module in src_conf: module_id = src_conf.get(module).get('module_id') if module_id is None: msg = "[error][file:%s][%s] module_id is null!" % (src, module) raise ParserError(msg) module_id = fn_str_to_int(module_id) if type(module_id) is not int else module_id if module_id > g_u8_max : msg = "[error][file:%s][%s] module_id is more than 0xFF!" % (src, module) raise ParserError(msg) # diffrent module must config with diffrent module_id if module_id in module_dict: if module_dict[module_id] != module: msg = "[error][file:%s][%s] module_id is the same to [%s]!" % (src, module, module_dict[module_id]) raise ParserError(msg) else: module_dict[module_id] = module if module not in combination: combination[module] = {'module_id': module_id} elif combination.get(module).get('module_id') != module_id: msg = "[error][%s][%s] module_id is not same as other file!" % (src, module) raise ParserError(msg) for item in src_conf.get(module): key_cfg = src_conf.get(module).get(item) if item == 'module_id': continue key_id = key_cfg.get('key_id') key_id = None if key_id is None else (fn_str_to_int(key_id) if type(key_id) is not int else key_id) # print("key id: %d, module id: %d, key>>8: %d"%( key_id, module_id, key_id>>8)) if key_id is None or key_id > g_u16_max : msg = "[error][file:%s][%s][%s] key_id is null or more than unsighed 16 or not match with module_id!" % (src, module, item) raise ParserError(msg) item_prod_type = key_cfg.get('product_type') key_status = key_cfg.get('key_status') #print('prodtype: %s, key prod: %s'%(prod_type, item_prod_type)) if (prod_type == item_prod_type or (item_prod_type is not None and prod_type in item_prod_type)) \ and key_status == 'alive': combination[module].update({item:key_cfg}) return combination def _nv_cfg_writer(self, dst_file, combination): if os.path.exists(os.path.dirname(dst_file)) is False: os.makedirs(os.path.dirname(dst_file)) if os.path.exists(dst_file): os.remove(dst_file) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL modes = stat.S_IWUSR | stat.S_IRUSR with os.fdopen(os.open(dst_file, flags, modes), 'w') as fout: fout.write(json.dumps(combination, indent=4)) def _merge_ver_cfg(self, file_prefix, ver_cfg): srcs = ver_cfg.get('srcs') if srcs is None: msg = "[error] ver cfg file is null!" raise ParserError(msg) prod_type = ver_cfg.get('prod_type') combination = self._prod_type_filter(srcs, prod_type) dst_file = '%s.json' % file_prefix self._nv_cfg_writer(dst_file, combination) return { "merged_cfg" : dst_file, "prod_type" : prod_type} def _create_header(self): pass # 检查配置文件中是否有必须配置项 def check_key(json_conf): check_item = ['OUT_BIN_DIR', 'BUILD_TEMP_PATH', 'NV_TARGET_JSON_PATH', \ 'NV_RELATIVE_PATH', 'NV_DEFAULT_CFG_DIR', 'DATABASE_TXT_FILE'] keys = dict.keys(json_conf) for check_key in check_item: if check_key not in keys: msg = "[error] [nv_binary] need add ConfigMap (%s) in json_conf!" % (check_key) raise ParserError(msg) def test(targets, flag, backup, use_crc16): root = g_root nv_target_json_path = os.path.join(root, json_conf["NV_TARGET_JSON_PATH"]) alias_conf = BuildConfParser(nv_target_json_path).get_conf_data() worker = BuildNv(alias_conf, root, targets, backup, use_crc16) if flag: worker.set_nv_output_dir(os.path.join(root, json_conf["OUT_BIN_DIR"])) worker.start_work() def nv_begin(in_path, targets, flag, gen_backup=False, use_crc16=False): global json_conf with open(in_path, 'r') as i: json_conf = json.load(i) check_key(json_conf) test(targets, flag, False, use_crc16) if gen_backup: test(targets, flag, True, use_crc16) print("build nv bin success!!") if __name__ == "__main__": in_path = sys.argv[2] targets = sys.argv[3].split() flag = len(sys.argv) == 3 nv_begin(in_path, targets, flag)