KnowledgeBase.git

			@@ -134,6 +134,15 @@
			"""
			return doc_dbh.get_text_with_entities(entity_names)

			@staticmethod
			def get_texts_with_entity(entity_names: list[str]) -> list[str]:
			"""
			根据实体词获取文档文本
			:param entity_names: str - 实体词名称
			:return: str - 文本内容
			"""
			return doc_dbh.get_texts_with_entities(entity_names)

			def run(self):
			# 根据文档，生成结构化数据
			self.handle_tm_structured_data()
			@@ -349,6 +358,7 @@
			def validation(gen_text):
			vcs = json.loads(gen_text)
			assert next(filter(lambda it: re.match('^[0-1]+$', it['VCID']), vcs)), '生成的VCID必须是二进制'

			doc_text = self.get_text_with_entity(['虚拟信道定义'])
			result = self.call_model(_msg, 'out/' + dev.code + '_虚拟信道.json', doc_text, validation)
			Log.info('虚拟信道：' + result)
			@@ -380,7 +390,8 @@
			pkts = json.loads(gen_text)
			assert len(pkts), 'VC源包列表不能为空'

			text = self.call_model(_msg, 'out/' + dev.code + '_遥测源包下传时机.json', ['遥测源包下传时机'], validation)
			doc_text = self.get_text_with_entity(['遥测源包下传时机'])
			text = self.call_model(_msg, 'out/' + dev.code + '_遥测源包下传时机.json', doc_text, validation)
			Log.info('遥测源包所属虚拟信道：' + text)
			return json.loads(text)

			@@ -410,7 +421,8 @@
			}
			]
			"""
			result = self.call_model(_msg, 'out/' + dev.code + '_源包列表.json', ['这里是文档中抽取的内容'])
			doc_text = self.get_text_with_entity(['源包列表'])
			result = self.call_model(_msg, 'out/' + dev.code + '_源包列表.json', doc_text)
			Log.info('遥测源包列表：' + result)
			return json.loads(result)

			@@ -434,7 +446,8 @@
			# 例子：
			{"last_par_pos":128, "par_num": 20}
			"""
			text = self.call_model(_msg, '', ['这里是文档中抽取的内容'])
			doc_text = self.get_text_with_entity([pkt_id])
			text = self.call_model(_msg, '', doc_text)
			result = json.loads(text)
			last_par_pos = result['last_par_pos']
			par_num = result['par_num']
			@@ -494,7 +507,7 @@
			]
			"""

			def validation(gen_text):
			def _validation(gen_text):
			_pkt = json.loads(gen_text)
			with open(f'out/tmp/{time.time()}.json', 'w') as f:
			f.write(gen_text)
			@@ -504,7 +517,7 @@
			# assert par_num == len(_pkt['datas']), f'数据域参数个数不对！预计{par_num}个，实际{len(_pkt["datas"])}'
			assert last_par_pos == _pkt['datas'][-1]['pos'], '最后一个参数的字节位置不对！'

			result = self.call_model(_msg, f'out/数据包-{pkt_name}.json', [], ['这里是文档中抽取的内容'], validation)
			result = self.call_model(_msg, f'out/数据包-{pkt_name}.json', doc_text, _validation)
			Log.info(f'数据包“{pkt_name}”信息：' + result)
			pkt = json.loads(result)
			else:
			@@ -517,87 +530,72 @@
			return pkt

			def gen_bus(self):
			_msg = """
			# 指令
			我需要从文档中提取经总线的数据包列表，你要帮助我完成经总线的数据包列表的提取。
			# 需求
			请析文档，列出总线通信包传输约定中描述的所有数据包列表，
			数据包字段包括：id(数据包代号)、name(数据包名称)、apid(16进制字符串)、service(服务子服务)、length(bit长度)、interval(传输周期)、subAddr(子地址/模式)、frameNum(通信帧号)、
			transSer(传输服务)、note(备注)、rtAddr(所属RT的地址十进制)、rt(所属rt名称)、throughBus(是否经过总线)、burst(是否突发)、transDirect(传输方向)。
			# 约束
			- frameNum：使用文档中的文本不要做任何转换；
			- subAddr：值为“深度”、“平铺”、“数字”或null；
			- 是否经过总线的判断依据：“备注”列填写了内容类似“不经过总线”的文字表示不经过总线否则经过总线；
			- 传输服务分三种：SetData(置数)、GetData(取数)、DataBlock(数据块传输)；
			- 传输方向分”收“和”发“，传输服务如果是”取数“是”收“，如果是”数据块传输“则根据包所在的分系统以及表格的”传输方向“列进行判断，判断对于SMU来说是收还是发；
			- 是否突发：根据表格中的”传输周期“列进行判断，如果填写了类似”突发“的文字表示是突发否则表示不是突发；
			- 不要漏掉任何一个数据包；
			- 数据结构最外层是数组，数组元素为数据包，以JSON格式输出，不要输出JSON以外的任何文本。
			# 例子
			[
			{
			"id": "PCS005",
			"name": "总线管理（内部指令）",
			"apid": "418",
			"service": "(1, 2)",
			"length": 1,
			"interval": 1000,
			"subAddr": null,
			"frameNum": "1\|2",
			"transSer": "DataBlock",
			"note": "",
			"rtAddr": 28,
			"rt": "数据接口单元XIU",
			"throughBus": true,
			"burst": true,
			"transDirect": "发"
			}
			]
			"""
			self.bus_pkts = []
			doc_text_list = self.get_texts_with_entity(['分系统源包'])
			for doc_text in doc_text_list:
			_msg = """
			# 指令
			我需要从文档中提取经总线的数据包列表，你要帮助我完成经总线的数据包列表的提取。
			# 需求
			请析文档，列出总线通信包传输约定中描述的所有数据包列表，
			数据包字段包括：id(数据包代号)、name(数据包名称)、apid(16进制字符串)、service(服务子服务)、length(bit长度)、interval(传输周期)、subAddr(子地址/模式)、frameNum(通信帧号)、
			transSer(传输服务)、note(备注)、rtAddr(所属RT的地址十进制)、rt(所属rt名称)、throughBus(是否经过总线)、burst(是否突发)、transDirect(传输方向)。
			# 约束
			- frameNum：使用文档中的文本不要做任何转换；
			- subAddr：值为“深度”、“平铺”、“数字”或null；
			- 是否经过总线的判断依据：“备注”列填写了内容类似“不经过总线”的文字表示不经过总线否则经过总线；
			- 传输服务分三种：SetData(置数)、GetData(取数)、DataBlock(数据块传输)；
			- 传输方向分”收“和”发“，传输服务如果是”取数“是”收“，如果是”数据块传输“则根据包所在的分系统以及表格的”传输方向“列进行判断，判断对于SMU来说是收还是发；
			- 是否突发：根据表格中的”传输周期“列进行判断，如果填写了类似”突发“的文字表示是突发否则表示不是突发；
			- 不要漏掉任何一个数据包；
			- 数据结构最外层是数组，数组元素为数据包，以JSON格式输出，不要输出JSON以外的任何文本。
			# 例子
			[
			{
			"id": "PCS005",
			"name": "总线管理（内部指令）",
			"apid": "418",
			"service": "(1, 2)",
			"length": 1,
			"interval": 1000,
			"subAddr": null,
			"frameNum": "1\|2",
			"transSer": "DataBlock",
			"note": "",
			"rtAddr": 28,
			"rt": "数据接口单元XIU",
			"throughBus": true,
			"burst": true,
			"transDirect": "发"
			}
			]
			"""

			def validation(gen_text):
			json.loads(gen_text)
			def validation(gen_text):
			json.loads(gen_text)

			result = self.call_model(_msg, 'out/总线.json', ['这里是文档中抽取的内容'], validation)
			Log.info('总线数据包：' + result)
			result = self.call_model(_msg, 'out/总线.json', doc_text, validation)
			Log.info('总线数据包：' + result)

			pkts = json.loads(result)
			# 筛选经总线的数据包
			pkts = list(filter(lambda it: it['throughBus'], pkts))
			# 筛选有apid的数据包
			pkts = list(filter(lambda it: it['apid'], pkts))
			pkts = json.loads(result)
			# 筛选经总线的数据包
			pkts = list(filter(lambda it: it['throughBus'], pkts))
			# 筛选有apid的数据包
			pkts = list(filter(lambda it: it['apid'], pkts))

			pkts2 = []
			# todo 这一步应该通过数据库筛选，数据库中已经有所有遥测包以及遥测包对应的定义段落文本
			for pkt in pkts:
			if self.pkt_in_tm_pkts(pkt["name"]):
			pkts2.append(pkt)
			for pkt in pkts2:
			self.gen_pkt_details(pkt['name'], pkt['id'])
			_pkt = self.gen_pkt_details(pkt['name'], pkt['id'])
			if _pkt:
			pkt['children'] = []
			pkt['children'].extend(_pkt['datas'])
			pkt['length'] = _pkt['length']
			self.bus_pkts = pkts

			def pkt_in_tm_pkts(self, pkt_name):
			_msg = f"""
			# 指令
			我需要从文档中分析判读是否有某个遥测包的字段表描述，你要帮助我判断。
			# 问题
			文档中有遥测包“{pkt_name}”的字段表描述吗？
			注意：遥测包的字段表紧接着遥测包章节标题，如果章节标题后面省略了或者详见xxx则是没有字段表描述。
			# 约束
			- 根据文档内容输出；
			- 遥测包名称必须完全匹配；
			- 输出“无”或“有”，不要输出其他任何内容。
			# 例子
			有
			"""
			text = self.call_model(_msg, f'out/pkts/有无数据包-{pkt_name}.txt', ['这里是文档中抽取的内容'])
			Log.info(f'文档中有无“{pkt_name}”的字段描述：' + text)
			return text == '有'
			# pkts2 = []
			# todo 这一步应该通过数据库筛选，数据库中存储了每个数据包的代号实体
			# for pkt in pkts:
			# if self.pkt_in_tm_pkts(pkt["name"]):
			# pkts2.append(pkt)
			for pkt in pkts:
			self.gen_pkt_details(pkt['name'], pkt['id'])
			_pkt = self.gen_pkt_details(pkt['name'], pkt['id'])
			if _pkt:
			pkt['children'] = []
			pkt['children'].extend(_pkt['datas'])
			pkt['length'] = _pkt['length']
			self.bus_pkts.extend(pkts)

			# endregion 遥测-end

			@@ -642,7 +640,8 @@
			def validation(gen_text):
			json.loads(gen_text)

			text = self.call_model(_msg, 'out/tc_transfer_frame.json', ['这里是文档中抽取的内容'], validation)
			doc_text = self.get_text_with_entity(['遥控帧格式'])
			text = self.call_model(_msg, 'out/tc_transfer_frame.json', doc_text, validation)
			result: dict = json.loads(text)
			format_text = utils.read_from_file('tpl/tc_transfer_frame.json')
			format_text = utils.replace_tpl_paras(format_text, result)
			@@ -681,7 +680,8 @@
			def validation(gen_text):
			json.loads(gen_text)

			text = self.call_model(_msg, 'out/tc_transfer_pkt.json', ['这里是文档中抽取的内容'], validation)
			doc_text = self.get_text_with_entity(['遥控包格式'])
			text = self.call_model(_msg, 'out/tc_transfer_pkt.json', doc_text, validation)
			result = json.loads(text)

			format_text = utils.read_from_file('tpl/tc_pkt_format.json')
			@@ -691,25 +691,29 @@
			return pkt_format

			def gen_tc_transfer_pkts(self):
			_msg = '''
			# 指令
			分析文档列出所有的遥控源包。
			# 输出例子：
			[{
			"name": "xxx",
			"code":"pkt",
			"应用过程标识符":"0xAA",
			"服务类型":"0x1",
			"服务子类型":"0x2"
			}]
			'''
			doc_text_list = self.get_texts_with_entity(['APID分配'])
			pkts = []
			for doc_text in doc_text_list:
			_msg = '''
			# 指令
			分析文档列出所有的遥控源包。
			# 输出例子：
			[{
			"name": "xxx",
			"code":"pkt",
			"应用过程标识符":"0xAA",
			"服务类型":"0x1",
			"服务子类型":"0x2"
			}]
			'''

			def validation(gen_text):
			json.loads(gen_text)
			def validation(gen_text):
			json.loads(gen_text)

			text = self.call_model(_msg, 'out/tc_transfer_pkts.json', ['这里是文档中抽取的内容'], validation)
			Log.info('遥控包列表：' + text)
			return json.loads(text)
			text = self.call_model(_msg, 'out/tc_transfer_pkts.json', doc_text, validation)
			Log.info('遥控包列表：' + text)
			pkts.extend(json.loads(text))
			return pkts

			def gen_tc_pkt_details(self, pkt):
			tc_name = pkt['name']