import os
|
from datetime import datetime
|
|
from openai import OpenAI
|
from pathlib import Path
|
import re
|
import json
|
import copy
|
|
from datas import pkt_vc, pkt_datas, dev_pkt, proj_data
|
from db.db_generate import create_project, create_device, create_data_stream
|
from db.models import TProject, TDevice
|
|
BASE_URL = 'https://dashscope.aliyuncs.com/compatible-mode/v1'
|
API_KEY = 'sk-15ecf7e273ad4b729c7f7f42b542749e'
|
MODEL_NAME = 'qwen-long'
|
|
assistant_msg = """
|
# 角色
|
你是一个专业的文档通信分析师，擅长进行文档分析和通信协议分析，同时能够解析 markdown 类型的文档。拥有成熟准确的文档阅读与分析能力，能够妥善处理多文档间存在引用关系的复杂情况。
|
|
## 技能
|
### 技能 1：文档分析（包括 markdown 文档）
|
1. 当用户提供文档时，仔细阅读文档内容，严格按照文档中的描述提取关键信息，不得加入自己的回答或建议。
|
2. 分析文档的结构、主题和重点内容，同样只依据文档进行表述。
|
3. 如果文档间存在引用关系，梳理引用脉络，明确各文档之间的关联，且仅呈现文档中体现的内容。
|
|
|
### 技能 2：通信协议分析
|
1. 接收通信协议相关信息，理解协议的规则和流程，仅依据所给信息进行分析。
|
|
## 目标导向
|
1. 通过对文档和通信协议的分析，为用户提供清晰、准确的数据结构，帮助用户更好地理解和使用相关信息。
|
2. 以 JSON 格式组织输出内容，确保数据结构的完整性和可读性。
|
|
## 规则
|
1. 每一个型号都会有一套文档，需准确判断是否为同一个型号的文档后再进行整体分析。
|
2. 每次只分析同一个型号。
|
3. 大多数文档结构为：型号下包含设备，设备下包含数据流，数据流下包含数据帧，数据帧中有一块是包域，包域中会挂载各种类型的数据包。
|
4. 这些文档都是数据传输协议的描述，在数据流、数据帧、数据包等传输实体中都描述了各个字段的分布和每个字段的大小，且大小单位不统一，需理解这些单位，并将所有输出单位统一为 bits，统一使用length表示。
|
5. 如果有层级，使用树形 JSON 输出，子节点 key 使用children；需保证相同类型的数据结构统一，并且判断每个层级是什么类型，输出类型字段，类型字段的 key 使用 type ；例如当前层级为字段时使用：type:"field"；当前层级为设备时使用：type:"device"
|
6.名称相关的字段的 key 使用name；代号或者唯一标识相关的字段的key使用id；序号相关的字段的key使用number；其他没有举例的字段使用精简的翻译作为字段的key；
|
7.探测帧为CADU，其中包含同步头和VCDU，按照习惯需要使用VCDU层级包含下一层级中传输帧主导头、传输帧插入域、传输帧数据域、传输帧尾的结构
|
|
## 限制：
|
- 所输出的内容必须按照JSON格式进行组织，不能偏离框架要求，且严格遵循文档内容进行输出，只输出 JSON ，不要输出其它文字。
|
- 不输出任何注释等描述性信息
|
|
"""
|
|
|
class DbStructFlow:
|
files = []
|
file_objects = []
|
|
def __init__(self, doc_files):
|
self.client = OpenAI(
|
api_key=API_KEY,
|
base_url=BASE_URL,
|
# api_key="ollama",
|
# base_url="http://192.168.1.48:11434/v1/",
|
)
|
if doc_files:
|
self.files = doc_files
|
self.load_file_objs()
|
self.delete_all_files()
|
self.upload_files()
|
|
def load_file_objs(self):
|
file_stk = self.client.files.list()
|
self.file_objects = file_stk.data
|
|
def delete_all_files(self):
|
for file_object in self.file_objects:
|
self.client.files.delete(file_object.id)
|
|
def upload_file(self, file_path):
|
file_object = self.client.files.create(file=Path(file_path), purpose="file-extract")
|
return file_object
|
|
def upload_files(self):
|
self.file_objects = []
|
for file_path in self.files:
|
file_object = self.upload_file(file_path)
|
self.file_objects.append(file_object)
|
|
def run(self):
|
# 生成型号结构
|
# 生成设备结构
|
# 生成数据流结构 CADU
|
# 生成VCDU结构
|
# 生成遥测数据包结构
|
proj = self.gen_project([])
|
# proj = TProject(C_PROJECT_PK='2e090a487c1a4f7f741be3a437374e2f')
|
|
devs = self.gen_device([], proj)
|
# with open('datas/设备列表.json', 'w', encoding='utf8') as f:
|
# json.dump(devs, f, ensure_ascii=False, indent=4)
|
#
|
# proj['devices'] = devs
|
#
|
# messages = []
|
# cadu = self.gen_tm_frame(messages)
|
# with open("datas/探测帧.json", 'w', encoding='utf8') as f:
|
# json.dump(cadu, f, ensure_ascii=False, indent=4)
|
#
|
# messages = []
|
# vcs = self.gen_vc(messages)
|
# with open('datas/虚拟信道.json', 'w', encoding='utf8') as f:
|
# json.dump(vcs, f, ensure_ascii=False, indent=4)
|
#
|
# messages = []
|
# pkt_vcs = self.gen_pkt_vc(messages)
|
# with open('datas/VC源包.json', 'w', encoding='utf8') as f:
|
# json.dump(pkt_vcs, f, ensure_ascii=False, indent=4)
|
#
|
# messages = []
|
# dev_pkts = self.gen_dev_pkts(messages)
|
# with open('datas/设备源包.json', 'w', encoding='utf8') as f:
|
# json.dump(dev_pkts, f, ensure_ascii=False, indent=4)
|
#
|
# messages = []
|
# _pkts = self.gen_pkts()
|
# pkts = []
|
# for pkt in _pkts:
|
# _pkt = self.gen_pkt_details(pkt['name'])
|
# pkts.append(_pkt)
|
# with open('datas/源包列表.json', 'w', encoding='utf8') as f:
|
# json.dump(pkts, f, ensure_ascii=False, indent=4)
|
#
|
# for dev in devs:
|
# ds = dev['data_streams'][0]
|
# _cadu = copy.deepcopy(cadu)
|
# ds['cadu'] = _cadu
|
# _vcdu = next(filter(lambda it: it['name'] == '传输帧', _cadu['children']))
|
# vcdu_data = next(filter(lambda it: it['name'] == '传输帧数据域', _vcdu['children']))
|
# _vcs = copy.deepcopy(vcs)
|
# vcdu_data['children'] = _vcs
|
# dev_pkt = next(filter(lambda it: it['name'] == dev['name'], dev_pkts), None)
|
# if dev_pkt is None:
|
# continue
|
# for pkt in dev_pkt['pkts']:
|
# for vc in _vcs:
|
# _pkt = next(
|
# filter(lambda it: it['name'] == pkt['name'] and it['vcs'].__contains__(vc['code']), pkt_vcs),
|
# None)
|
# if _pkt:
|
# if vc.__contains__('pkts') is False:
|
# vc['pkts'] = []
|
# _pkt = next(filter(lambda it: it['name'] == _pkt['name'], pkts), None)
|
# if _pkt:
|
# vc['pkts'].append(_pkt)
|
#
|
# with open("datas/型号.json", 'w', encoding='utf8') as f:
|
# json.dump(proj, f, ensure_ascii=False, indent=4)
|
return ''
|
|
def _gen(self, msgs, msg):
|
messages = [] if msgs is None else msgs
|
if len(messages) == 0:
|
# 如果是第一次提问加入文档
|
messages.append({'role': 'system', 'content': assistant_msg})
|
for file_object in self.file_objects:
|
messages.append({'role': 'system', 'content': 'fileid://' + file_object.id})
|
messages.append({'role': 'user', 'content': msg})
|
|
completion = self.client.chat.completions.create(
|
model=MODEL_NAME,
|
messages=messages,
|
stream=True,
|
temperature=0.0,
|
top_p=0,
|
timeout=30 * 60000,
|
max_completion_tokens=1000000
|
# stream_options={"include_usage": True}
|
)
|
|
text = ''
|
for chunk in completion:
|
if chunk.choices[0].delta.content is not None:
|
text += chunk.choices[0].delta.content
|
print(chunk.choices[0].delta.content, end="")
|
print("")
|
return text
|
|
def gen_project(self, messages):
|
_msg = f"""
|
根据文档输出型号信息，型号字段包括：名称和代号，仅输出型号的属性，不输出其他层级数据
|
"""
|
print('型号信息：')
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
proj_dict = json.loads(text)
|
# return proj_dict
|
code = proj_dict['id']
|
name = proj_dict['name']
|
proj = create_project(code, name, code, name, "", datetime.now())
|
return proj
|
|
def gen_device(self, messages, proj):
|
"""
|
设备列表生成规则：
|
1.如文档中有1553协议描述，加入1553设备
|
2.如是类SMU软件（遥测遥控包含，BC或者RT），加入对应相关设备，文档只有设备名称和设备ID，设备类型90%是标准类型
|
3.如是类RTU软件，加入对应相关设备，文档里面有设备名称和设备ID，同上
|
4.如基于软平台，如是SMU软件，加入SMU工控机设备，待定
|
|
设备类型：工控机[0]、1553B[1]
|
|
:param messages:
|
:return:
|
"""
|
proj_pk = proj.C_PROJECT_PK
|
devices = []
|
|
_msg = f"""
|
输出所有设备列表，设备字段包括名称（name)、代号（code），如果没有代号则使用名称的英文翻译缩写代替且缩写长度不超过5个字符，JSON格式，并且给每个设备增加三个字段，第一个字段hasTcTm“是否包含遥控遥测”，判断该设备是否包含遥控遥测的功能；第二个字段hasTemperatureAnalog“是否包含温度量、模拟量等数据的采集”，判断该设备是否包含温度量等信息的采集功能；第三个字段hasBus“是否是总线设备”，判断该设备是否属于总线设备，是否有RT地址；每个字段的值都使用true或false来表示。
|
仅输出JSON，不要输出JSON以外的任何字符。
|
"""
|
print('设备列表：')
|
text = self._gen(messages, _msg)
|
text = self.remove_markdown(text)
|
devs = json.loads(text)
|
hasBus = any(d['hasBus'] for d in devs)
|
if hasBus:
|
# 总线设备
|
dev = create_device("B1553", "1553总线", '1', 'StandardProCommunicationDev', proj_pk)
|
devices.append(dev)
|
# 创建数据流
|
ds_u153 = create_data_stream(proj_pk, dev.C_DEV_PK, 'ECSS上行总线数据', 'U153', 'B153', '0', 'E153', '001')
|
ds_d153 = create_data_stream(proj_pk, dev.C_DEV_PK, 'ECSS下行总线数据', 'D153', 'B153', '1', 'E153', '001')
|
|
# 类SMU设备，包含遥测和遥控功能，名称结尾为“管理单元”
|
like_smu_devs = list(filter(lambda it: it['hasTcTm'] and it['name'].endswith('管理单元'), devs))
|
for dev in like_smu_devs:
|
dev = create_device(dev['code'], dev['name'], '0', 'StandardProCommunicationDev', proj.C_PROJECT_PK)
|
devices.append(dev)
|
# 创建数据流
|
ds_tmfl = create_data_stream(proj_pk, dev.C_DEV_PK, 'AOS遥测', 'TMFL', 'TMFL', '1', 'TMFL', '001')
|
ds_tcfl = create_data_stream(proj_pk, dev.C_DEV_PK, '遥控指令', 'TCFL', 'TCFL', '0', 'TCFL', '006')
|
|
# 类RTU设备，包含温度量和模拟量功能，名称结尾为“接口单元”
|
like_rtu_devs = list(filter(lambda it: it['hasTemperatureAnalog'] and it['name'].endswith('接口单元'), devs))
|
for dev in like_rtu_devs:
|
dev = create_device(dev['code'], dev['name'], '0', 'StandardProCommunicationDev', proj.C_PROJECT_PK)
|
|
# for dev in like_rtu_devs:
|
# dev = create_device(dev['code'], dev['name'], '0', '', proj.C_PROJECT_PK)
|
# devices.append(dev)
|
# # 创建数据流
|
# ds_tmfl = create_data_stream(proj.C_PROJECT_PK, '温度量', 'TMFL', 'TMFL', '1', 'TMFL', '001')
|
# ds_tcfl = create_data_stream(proj.C_PROJECT_PK, '模拟量', 'TCFL', 'TCFL', '0', 'TCFL', '006')
|
|
print()
|
# 总线设备
|
# print('是否有总线设备：', end='')
|
# _msg = "文档中描述的有总线相关内容吗？仅回答：“有”或“无”，不要输出其他文本。"
|
# text = self._gen([], _msg)
|
# if text == "有":
|
# _msg = f"""
|
# 文档中描述的总线型号是多少，仅输出总线型号不要输出型号以外的其他任何文本，总线型号由数字和英文字母组成。
|
# """
|
# print('设备ID：')
|
# dev_code = self._gen([], _msg)
|
# dev = create_device(dev_code, dev_code, '1', '', proj.C_PROJECT_PK)
|
# devices.append(dev)
|
|
# 类SMU软件
|
# print('是否有类SMU设备：', end='')
|
# _msg = "文档中有描述遥测和遥控功能吗？仅回答：“有”或“无”，不要输出其他文本。"
|
# text = self._gen([], _msg)
|
# if text == "有":
|
# # 系统管理单元
|
# print('是否有系统管理单元（SMU）：', end='')
|
# _msg = f"文档中有描述系统管理单元（SMU）吗？仅回答“有”或“无”，不要输出其他文本。"
|
# text = self._gen([], _msg)
|
# if text == "有":
|
# dev = create_device("SMU", "系统管理单元", '0', '', proj.C_PROJECT_PK)
|
# devices.append(dev)
|
# # 中心控制单元（CTU）
|
# print('是否有中心控制单元（CTU）：', end='')
|
# _msg = f"文档中有描述中心控制单元（CTU）吗？仅回答“有”或“无”，不要输出其他文本。"
|
# text = self._gen([], _msg)
|
# if text == "有":
|
# dev = create_device("CTU", "中心控制单元", '0', '', proj.C_PROJECT_PK)
|
# devices.append(dev)
|
#
|
# # 类RTU
|
# print('是否有类RTU设备：', end='')
|
# _msg = "文档中有描述模拟量采集和温度量采集功能吗？仅回答：“有”或“无”，不要输出其他文本。"
|
# text = self._gen([], _msg)
|
# if text == "有":
|
# dev = create_device("RTU", "远置单元", '0', '', proj.C_PROJECT_PK)
|
# devices.append(dev)
|
# device_dicts = json.loads(text)
|
# for device_dict in device_dicts:
|
# data_stream = {'name': '数据流', 'code': 'DS'}
|
# device_dict['data_streams'] = [data_stream]
|
#
|
# return device_dicts
|
return devices
|
|
def gen_tm_frame(self, messages):
|
_msg = f"""
|
输出探测帧的结构，探测帧字段包括：探测帧代号(id)、探测帧名称(name)、长度(length)、下级数据单元列表（children）。代号如果没有则用名称的英文翻译，包括下级数据单元。
|
"""
|
print('探测帧信息：')
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
cadu = json.loads(text)
|
|
return cadu
|
|
def gen_vc(self, messages):
|
_msg = f"""
|
输出探测虚拟信道的划分，不需要描述信息，使用一个数组输出，字段包括：代号(code)、vcid、名称(name)。
|
"""
|
print('虚拟信道：')
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
vcs = json.loads(text)
|
return vcs
|
|
def gen_dev_pkts(self, messages):
|
_msg = f"""
|
输出文档中探测源包类型定义描述的设备以及设备下面的探测包，数据结构：最外层为设备列表 > 探测包列表(pkts)，设备字段包括：名称(name)、代号(id)，源包字段包括：名称(name)、代号(id)
|
"""
|
print('设备探测源包信息：')
|
file = next(filter(lambda it: it.filename == 'XA-5D无人机分系统探测源包设计报告（公开）.md', self.file_objects),
|
None)
|
messages = [{'role': 'system', 'content': assistant_msg}, {'role': 'system', 'content': 'fileid://' + file.id}]
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
dev_pkts = json.loads(text)
|
return dev_pkts
|
|
def gen_pkt_details(self, pkt_name):
|
_msg = f"""
|
输出文档中描述的“{pkt_name}”探测包。
|
探测包字段包括：名称(name)、代号(id)、包头属性列表(headers)、数据域参数列表(datas)，
|
包头属性字段包括：位置(pos)、名称(name)、代号(id)、定义(val)，
|
数据域参数字段包括：位置(pos)、名称(name)、代号(id)、字节顺序(byteOrder)，
|
如果没有代号用名称的英文翻译代替，如果没有名称用代号代替，
|
输出内容仅输出json，不要输出任何其他内容！
|
"""
|
print(f'探测源包“{pkt_name}”信息：')
|
file = next(filter(lambda it: it.filename == 'XA-5D无人机分系统探测源包设计报告（公开）.md', self.file_objects),
|
None)
|
messages = [{'role': 'system', 'content': assistant_msg}, {'role': 'system', 'content': 'fileid://' + file.id}]
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
pkt = json.loads(text)
|
return pkt
|
|
def gen_pkts(self):
|
_msg = f"""
|
输出文档中描述的探测包。
|
探测包字段包括：名称(name)、代号(id)，
|
如果没有代号用名称的英文翻译代替，如果没有名称用代号代替，
|
顶级结构直接从探测包开始，不包括探测包下面的参数。
|
"""
|
print(f'探测源包列表：')
|
file = next(
|
filter(lambda it: it.filename == 'XA-5D无人机分系统探测源包设计报告（公开）.md', self.file_objects),
|
None)
|
messages = [{'role': 'system', 'content': assistant_msg},
|
{'role': 'system', 'content': 'fileid://' + file.id}]
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
pkt = json.loads(text)
|
return pkt
|
|
def gen_pkt_vc(self, messages):
|
_msg = f"""
|
根据探测源包下传时机定义，输出各个探测源包信息列表，顶级结构为数组元素为探测源包，源包字段包括：包代号(id)，名称(name)，所属虚拟信道(vcs)，下传时机（timeTags）
|
"""
|
print('探测源包所属虚拟信道：')
|
text = self._gen(messages, _msg)
|
messages.append({'role': 'assistant', 'content': text})
|
text = self.remove_markdown(text)
|
pkt_vcs = json.loads(text)
|
return pkt_vcs
|
|
def remove_markdown(self, text):
|
# 去掉开头的```json
|
text = re.sub(r'^```json', '', text)
|
# 去掉结尾的```json
|
text = re.sub(r'```$', '', text)
|
return text
|
|
|
if __name__ == '__main__':
|
md_file = 'D:\\workspace\\PythonProjects\\KnowledgeBase\\doc\\文档合并.md'
|
md_file2 = 'D:\\workspace\\PythonProjects\\KnowledgeBase\\doc\\XA-5D无人机分系统探测源包设计报告（公开）.md'
|
# 启动大模型处理流程
|
ret_text = DbStructFlow([md_file, md_file2]).run()
|
