| | |
|---|
| | | 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 |
|---|
| | |
|---|
| | | - 不输出任何注释等描述性信息。 |
|---|
| | | """ |
|---|
| | | |
|---|
| | | tc_system_msg = """ |
|---|
| | | # 角色 |
|---|
| | | 你是一个资深软件工程师。 |
|---|
| | | # 约束 |
|---|
| | | - 输出内容必须根据文档和问题回答，不要创造其他内容； |
|---|
| | | - 输出内容必须是，JSON格式，不要输出其他文本。 |
|---|
| | | """ |
|---|
| | | |
|---|
| | | g_completion = None |
|---|
| | | |
|---|
| | | |
|---|
| | | def read_from_file(cache_file): |
|---|
| | | with open(cache_file, 'r', encoding='utf-8') as f: |
|---|
| | |
|---|
| | | |
|---|
| | | self.gen_tc() |
|---|
| | | return '' |
|---|
| | | |
|---|
| | | |
|---|
| | | def get_text_with_entity(self, entity_names: list[str]) -> str: |
|---|
| | | """ |
|---|
| | | 根据实体词获取文档文本 |
|---|
| | |
|---|
| | | 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): |
|---|
| | |
|---|
| | | 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) |
|---|
| | | |
|---|
| | |
|---|
| | | |
|---|
| | | def gen_pkt_details(self, pkt_name, pkt_id): |
|---|
| | | cache_file = f'out/数据包-{pkt_name}.json' |
|---|
| | | |
|---|
| | | |
|---|
| | | # _msg = f""" |
|---|
| | | # #角色 |
|---|
| | | # 你是一名资深的软件工程师。 |
|---|
| | |
|---|
| | | # #例子： |
|---|
| | | # {"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'] |
|---|
| | |
|---|
| | | 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 |
|---|
| | |
|---|
| | | ] |
|---|
| | | """ |
|---|
| | | 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 |
|---|
| | |
|---|
| | | 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) |
|---|
| | |
|---|
| | | # 遥控包格式 |
|---|
| | | 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 |
|---|
| | | |
|---|
| | |
|---|
| | | 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']: |
|---|
| | |
|---|
| | | 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): |
|---|
| | | """ |
|---|
| | |
|---|
| | | 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: |
|---|
| | |
|---|
| | | - 通过标志：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文本，不要输出任何其他文本。 |
|---|
| | |
|---|
| | | - 命令正确应答: 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文本，不要输出任何其他文本。 |
|---|
| | |
|---|
| | | # 指令 |
|---|
| | | 分析文档列出所有的遥控指令。 |
|---|
| | | # 约束 |
|---|
| | | - 应用过程标识：应用过程标识就是APID，一般会在名称后的括号中列出来 |
|---|
| | | - code是指令代号，没有就空着 |
|---|
| | | - 应用过程标识：应用过程标识就是APID，一般会在名称后的括号中列出来； |
|---|
| | | - code：指令代号，没有就空着； |
|---|
| | | - name：指令名称，根据表格内容提取，注意名称需要提取完整，如果有多列则合并用-分割； |
|---|
| | | - 应用数据区：提取表格中的应用数据区内容。 |
|---|
| | | # 输出例子： |
|---|
| | | [{ |
|---|
| | | "name": "xxx", |
|---|
| | | "code":"pkt", |
|---|
| | | "应用过程标识符":"0xAA", |
|---|
| | | "服务类型":"0x1", |
|---|
| | | "服务子类型":"0x2" |
|---|
| | | "服务子类型":"0x2", |
|---|
| | | "应用数据区": "" |
|---|
| | | }] |
|---|
| | | ''' |
|---|
| | | |
|---|
| | |
|---|
| | | |
|---|
| | | # 指令 |
|---|
| | | 分析文档，从文档中提取遥控指令名称为“{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。 |
|---|
| | | 计算公式定义： |
|---|
| | | - 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，长度包括start和end，必须使用长度描述中的字段； |
|---|
| | | - 输出数据结构为数组，数组元素为字段信息； |
|---|
| | | - 输出内容必须为严格的json，不能输出除json以外的任何内容。 |
|---|
| | | |
|---|
| | | # 输出例子： |
|---|
| | | [ |
|---|
| | |
|---|
| | | "length": 8, |
|---|
| | | "type": "const", |
|---|
| | | "value": "0xAA" |
|---|
| | | }, |
|---|
| | | { |
|---|
| | | "name": "para2", |
|---|
| | | "code": "para2", |
|---|
| | | "length": 8, |
|---|
| | | "type": "length", |
|---|
| | | "value": {"start": "data", "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) |
|---|
