lyg
2025-05-22 c099e6662b8a6e320ac314d31eda9b40455e5aa7
db_struct_flow.py
@@ -12,7 +12,7 @@
import data_templates
from knowledgebase import utils
from knowledgebase.db.db_helper import create_project, create_device, create_data_stream, \
    update_rule_enc, create_extend_info, create_ref_ds_rule_stream, create_ins_format
    update_rule_enc, create_extend_info, create_ref_ds_rule_stream, create_ins_format, make_attr
from knowledgebase.db.data_creator import create_prop_enc, create_enc_pkt, get_data_ty, create_any_pkt
from knowledgebase.db.models import TProject
@@ -111,7 +111,16 @@
- 不输出任何注释等描述性信息。
"""
tc_system_msg = """
# 角色
你是一个资深软件工程师。
# 约束
- 输出内容必须根据文档和问题回答,不要创造其他内容;
- 输出内容必须是,JSON格式,不要输出其他文本。
"""
g_completion = None
def read_from_file(cache_file):
    with open(cache_file, 'r', encoding='utf-8') as f:
@@ -345,7 +354,7 @@
        self.gen_tc()
        return ''
    def get_text_with_entity(self, entity_names: list[str]) -> str:
        """
        根据实体词获取文档文本
@@ -416,6 +425,9 @@
        return self.generate_text(msg, cache_file, msgs, doc_text, validation, try_cnt, True)
    def generate_tc_text(self, msg, cache_file, msgs=None, doc_text=None, validation=None, try_cnt=5):
        msgs = [
            {'role': 'system', 'content': tc_system_msg},
            {'role': 'user', 'content': "以下是文档内容:\n" + doc_text}]
        return self.generate_text(msg, cache_file, msgs, doc_text, validation, try_cnt, True)
    def gen_project(self):
@@ -491,7 +503,7 @@
            assert isinstance(_devs, list), '数据结构最外层不是数组'
            assert next(filter(lambda it: it['name'].endswith('管理单元'), _devs), None), '生成的设备列表中没有管理单元'
        doc_text = self.get_text_with_entity(['系统概述','总线管理'])
        doc_text = self.get_text_with_entity(['系统概述', '总线管理'])
        text = self.generate_text_json(_msg, cache_file, doc_text=doc_text, validation=validation)
        devs = json.loads(text)
@@ -746,7 +758,7 @@
    def gen_pkt_details(self, pkt_name, pkt_id):
        cache_file = f'out/数据包-{pkt_name}.json'
        # _msg = f"""
        #     #角色
        #     你是一名资深的软件工程师。
@@ -765,7 +777,7 @@
        #     #例子:
        #     {"last_par_pos":128, "par_num": 20}
        #     """
        # text = self.generate_text_json(_msg, '', doc_text=doc_text)
        # result = json.loads(text)
        # last_par_pos = result['last_par_pos']
@@ -843,7 +855,7 @@
            return None
        text = self.generate_text_json(_msg, cache_file, [], doc_text, validation)
        pkt = json.loads(text)
        pkt_len = 0
        for par in pkt['datas']:
            par['pos'] = pkt_len
@@ -905,11 +917,12 @@
]
        """
        print('遥测源包所属虚拟信道:')
        def validation(gen_text):
            pkts = json.loads(gen_text)
            assert len(pkts), 'VC源包列表不能为空'
        doc_text = self.get_text_with_entity(['虚拟信道定义','遥测源包下传时机'])
        doc_text = self.get_text_with_entity(['虚拟信道定义', '遥测源包下传时机'])
        text = self.generate_text_json(_msg, 'out/遥测VC源包.json', doc_text=doc_text, validation=validation)
        pkt_vcs = json.loads(text)
        return pkt_vcs
@@ -1015,9 +1028,7 @@
        def validation(gen_text):
            json.loads(gen_text)
        doc_text = self.get_text_with_entity(['RT地址分配','分系统源包'])
        doc_text = self.get_text_with_entity(['RT地址分配', '分系统源包'])
        text = self.generate_text_json(_msg, 'out/总线.json', doc_text=doc_text,
                                       validation=validation)
        pkts = json.loads(text)
@@ -1118,18 +1129,18 @@
        # 遥控包格式
        pkt_format = self.gen_tc_pkt_format()
        # 遥控包列表
        pkts = self.gen_tc_transfer_pkts()
        for pkt in pkts:
            # 遥控包数据区内容
            self.gen_tc_pkt_details(pkt)
            pkt['type'] = 'insUnit'
        instructions = self.gen_tc_transfer_pkts()
        for inst in instructions:
            # 遥控指令数据区内容
            self.gen_tc_pkt_details(inst)
            inst['type'] = 'insUnit'
            format_text = json.dumps(pkt_format, ensure_ascii=False)
            format_text = utils.replace_tpl_paras(format_text, pkt)
            format_text = utils.replace_tpl_paras(format_text, inst)
            pf = json.loads(format_text)
            pf['name'] = pkt['name']
            pf['code'] = pkt['code']
            pf['name'] = inst['name']
            pf['code'] = inst['code']
            data_area = next(filter(lambda x: x['name'] == '应用数据区', pf['children']))
            data_area['children'].append(pkt)
            data_area['children'].append(inst)
            frame['subPkts'].append(pf)
        self.order = 0
@@ -1143,7 +1154,7 @@
            elif item['type'] == 'length':
                return None
            elif item['type'] == 'checkSum':
                return json.dumps({"ChecksumType": "CRC-CCITT"})
                return json.dumps({"ChecksumType": item['value']['type']})
            elif item['type'] == 'subPkt':
                return json.dumps({"CanInput": False})
            elif item['type'] in ['combPkt', 'insUnitList', 'input']:
@@ -1152,26 +1163,10 @@
                return '{"MinLength":null,"MaxLength":null,"IsSubPackage":false,"InputParams":[],"OutPutParams":[],"MatchItems":[]}'
            elif item['type'] == 'pkt':
                return '''{"MaxLength":1024,"IsSplit8":false,"Split8Start":null,"Split8End":null,"PadCode":null,"Alignment":null,"InputParams":[],"OutPutParams":[],"MatchItems":[]}'''
            elif item['type'] == 'pktSeqCnt':
                return json.dumps({"FirstPackValue":"PackCount","MiddlePackValue":"PackIndex","LastPackValue":"PackIndex","IndependPackValue":"InsUnitCount"})
            elif 'value' in item:
                return item['value']
        def make_attr(ty: str):
            """
            获取字段定义的ATTR。
            位掩码,用于标识节点类型。
            类型:0~2 BinaryType;
                3~5 DataType;
                6~8: InputFormat;
                9 : IsSubPackage;
                10: IsSendFlag;
                11~13: ProcessMethod;
                14~16: ExpressionType;
                17~19: EnumType
            :param ty:
            :return:
            """
        def create_tc_format(parent_pk, field, parent_parent_pk=None):
            """
@@ -1199,11 +1194,20 @@
            if 'length' in field:
                field['bitWidth'] = field['length']
            field['bitOrder'] = None
            field['attr'] = 0
            field['attr'] = make_attr(field)
            if field['type'] == 'length' and 'value' in field and field['value']:
                val = field['value']
                field['range'] = val['start'] + "~" + val['end']
                field['formula'] = val['formula']
            # 即时输入长度为null则是变长字段,需要把类型改为variableLength
            if field['type'] == 'input' and field['length'] is None:
                field['type'] = 'variableLength'
            # 枚举值默认值设置
            if field['type'] == 'enum' and len(field['enums']) and not next(filter(lambda x: 'default' in x and x['default'], field['enums']), None):
                field['enums'][0]['default'] = True
            # 校验和
            if field['type'] == 'checkSum':
                field['range'] = f'{field["value"]["start"]}~{field["value"]["end"]}'
            ins_format = create_ins_format(self.proj.C_PROJECT_PK, parent_pk, field)
            ins_format_pk = ins_format.C_INS_FORMAT_PK
            if 'children' in field:
@@ -1242,11 +1246,12 @@
- 通过标志:const,二进制,以B结尾;
- 控制命令标志:const,二进制,以B结尾;
- 空闲位:const,二进制,以B结尾;
- 航天器标识:const,十六进制,以0x开头;
- 航天器标识:const,十六进制,以0x开头,如果是二进制或十进制需要转换为十六进制;
- 虚拟信道标识:sendFlag,发送标记,默认为“任务注入帧”,所有的值都要列举出来;
# 数据类型
- const:固定码字,数值,二进制以B结尾,十进制,十六进制以0x开头;
- sendFlag:发送标记,类似枚举,定义样例:[{"n":"name","v":"value","c":"code","default":true}],n表示名称,v表示值,c表示code(没有空着),default表示是默认值;
- checkSum:校验和,如果是校验和类型还需要分析校验和的算法,并保存在value中,校验和算法包括:字节异或(ByteXOR)、累加和取反(SumNot)、累加和(AddSum)、应用循环冗余(CRC-CCITT)、CRC8(CRC8)、ISO和校验(ISOSum)、奇校验(Odd)、偶校验(Even)、其他(Other)
# 约束
- 以JSON格式输出;
- 仅输出JSON文本,不要输出任何其他文本。
@@ -1288,7 +1293,19 @@
- 命令正确应答: const,二进制;
- 源地址: const,十六进制。
# 数据类型
- const:固定码字,数值,二进制以B结尾,十进制,十六进制以0x开头;
- 固定码字:const,数值,二进制以B结尾,十进制,十六进制以0x开头;
- 长度:length,如果字段描述内容为数据区域的长度则表示是长度,长度的value为数值、null或范围定义,
- 枚举值:enum,
- 校验和:checkSum,如果是校验和类型还需要分析校验和的算法,并保存在value中,校验和算法包括:字节异或(ByteXOR)、累加和取反(SumNot)、累加和(AddSum)、应用循环冗余(CRC-CCITT)、CRC8(CRC8)、ISO和校验(ISOSum)、奇校验(Odd)、偶校验(Even)、其他(Other)
- 即时输入:input。
# 长度类型的范围定义描述
{"start": "起始字段code", "end": "结束字段code", "formula": "计算公式"}
- start:起始字段code,长度包括起始字段,字段描述中说明了起始字段,
- end:结束字段code,长度包括结束字段,字段描述中说明了结束字段,
- formula:计算公式,如果没有计算相关描述则表示不需要计算公式。
计算公式定义:
- BYTES:按字节计算;
- N-x:总字节数减x,例如总字节数减1的公式为N-1。
# 约束
- 以JSON格式输出;
- 仅输出JSON文本,不要输出任何其他文本。
@@ -1320,15 +1337,17 @@
# 指令
分析文档列出所有的遥控指令。
#  约束
- 应用过程标识:应用过程标识就是APID,一般会在名称后的括号中列出来
- code是指令代号,没有就空着
- 应用过程标识:应用过程标识就是APID,一般会在名称后的括号中列出来;
- code:指令代号,没有就空着;
- 应用数据区:提取表格中的应用数据区内容。
# 输出例子:
[{
"name": "xxx",
"code":"pkt",
"应用过程标识符":"0xAA",
"服务类型":"0x1",
"服务子类型":"0x2"
"服务子类型":"0x2",
"应用数据区": ""
}]
'''
@@ -1352,28 +1371,43 @@
# 指令
分析文档,从文档中提取遥控指令名称为“{tc_name}”代号为“{tc_code}”的指令应用数据区定义。
有些文档内容非常简单仅仅包含特定字节的内容描述,如果是这种文档,则每个特定字节的内容描述定义为一个字段,字段类型根据字节内容确定。
""" + """
# 约束
- code 如果没有明确定义则使用名称的英文翻译,尽量简短;
- length 自动转换为bit长度,必须是数值或null,不能为0;
- value 根据字段描述提取;
- enums 有些字段是枚举值,根据字段描述提取,枚举元素的数据结构为{"n":"","v":"","c":""};
- 输出内容必须为严格的json,不能输出除json以外的任何内容。
# 字段类型
- 固定码字:const,
- 长度:length,
- 固定码字:const,数值,二进制以B结尾,十进制,十六进制以0x开头;
- 长度:length,如果字段描述内容为数据区域的长度则表示是长度,长度的value为数值、null或范围定义,
- 枚举值:enum,
- 校验和:checkSum,
- 即时输入:input。
- 校验和:checkSum,如果是校验和类型还需要分析校验和的算法是什么,并保存在value中,
- 即时输入:input,如果是即时输入value的值为空字符串。
# 长度类型的范围定义描述
{"start": "起始字段code", "end": "结束字段code", "formula": "计算公式"}
- start:起始字段code,长度包括起始字段,字段描述中说明了起始字段,
- end:结束字段code,长度包括结束字段,字段描述中说明了结束字段,
- formula:计算公式,如果没有计算相关描述则表示不需要计算公式。
计算公式定义:
- BYTES:按字节计算;
- N-x:总字节数减x,例如总字节数减1的公式为N-1。
# 字段类型分析方法
- 根据字段描述分析字段的类型;
- 字段描述中明确指定了字段值的,类型为const;
- 字段中没有明确指定字段值,但是罗列了取值范围的,类型为enum;
- 字段描述中没有明确指定字段值,但是罗列了取值范围的,类型为enum;
- 字段描述中如果没有明确指定字段值也没有罗列取值范围的,类型为input;
- 字段如果是和“长度”有关,类型为length;
- 如果和数据域有关,类型为const;
- 字段如果和校验和有关,类型为checkSum。
- 字段如果和校验和有关,类型为checkSum,分析校验和的算法,并保存在value中,校验和算法包括:字节异或(ByteXOR)、累加和取反(SumNot)、累加和(AddSum)、应用循环冗余(CRC-CCITT)、CRC8(CRC8)、ISO和校验(ISOSum)、奇校验(Odd)、偶校验(Even)、其他(Other)。
# 约束
- code 如果没有明确定义则使用名称的英文翻译,尽量简短;
- length 自动转换为bit长度,必须是数值、null或范围定义,不能为0;
- value 根据字段描述提取字段值,字段值一般为数值类型,需要根据字段类型来分析,如果是length类型value的值为范围定义;
- enums 枚举类型的字段必须要有enums,根据字段描述提取,枚举元素的数据结构为{"n":"","v":"","c":""};
- length类型的范围定义中的start和end必须是生成结果中的字段code;
- 输出数据结构为数组,数组元素为字段信息;
- 输出内容必须为严格的json,不能输出除json以外的任何内容。
# 输出例子:
[
@@ -1383,18 +1417,52 @@
        "length": 8,
        "type": "const",
        "value": "0xAA"
    },
    {
        "name": "para2",
        "code": "para2",
        "length": 8,
        "type": "length",
        "value": {"start": "para1", "end": "data", "formula": "BYTES"}
    },
    {
        "name": "数据",
        "code": "data",
        "length": null,
        "type": "input",
        "value": ""
    }
    ...
]
"""
        def validation(gen_text):
            json.loads(gen_text)
            fields = json.loads(gen_text)
            for field in fields:
                if field['type'] == 'length':
                    if field['value'] is None:
                        raise Exception('length类型的value不能为空')
                    if 'start' not in field['value'] or 'end' not in field['value']:
                        raise Exception('length类型的value必须包含start和end')
                    if field['value']['start'] not in [f['code'] for f in fields]:
                        raise Exception('length类型的value的start字段必须在fields中')
                    if field['value']['end'] not in [f['code'] for f in fields]:
                        raise Exception('length类型的value的end字段必须在fields中')
                elif field['type'] == 'enum':
                    if 'enums' not in field:
                        raise Exception('enum类型的field必须包含enums')
                    if len(field['enums']) == 0:
                        raise Exception('enum类型的field的enums不能为空')
                    for enum in field['enums']:
                        if 'n' not in enum or 'v' not in enum:
                            raise Exception('enum类型的field的enums的元素必须包含n、v、c')
                        if enum['n'] == '' or enum['v'] == '':
                            raise Exception('enum类型的field的enums的元素不能为空')
        doc_text = self.get_text_with_entity([tc_name])
        if doc_text == '':
            return None
        text = self.generate_tc_text(_msg, f'out/遥控指令数据域-{tc_code}-{utils.to_file_name(tc_name)}.json',
            doc_text = pkt['应用数据区']
        text = self.generate_tc_text(_msg,
                                     f'out/遥控指令数据域-{tc_code}-{utils.to_file_name(tc_name)}.json',
                                     doc_text=doc_text,
                                     validation=validation)
        result = json.loads(text)