668 lines
30 KiB
Python
668 lines
30 KiB
Python
|
#!/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)
|