mirror of
https://github.com/goldenfishs/MRobot.git
synced 2025-07-27 08:49:01 +08:00
806 lines
35 KiB
Python
806 lines
35 KiB
Python
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QStackedLayout, QFileDialog, QHeaderView
|
||
from PyQt5.QtCore import Qt
|
||
from PyQt5.QtWidgets import QTreeWidgetItem as TreeItem
|
||
from qfluentwidgets import TitleLabel, BodyLabel, SubtitleLabel, StrongBodyLabel, HorizontalSeparator, PushButton, TreeWidget, InfoBar,FluentIcon
|
||
import os
|
||
import requests
|
||
import zipfile
|
||
import io
|
||
import shutil
|
||
import yaml
|
||
from jinja2 import Template
|
||
|
||
class IocConfig:
|
||
def __init__(self, ioc_path):
|
||
self.ioc_path = ioc_path
|
||
self.config = {}
|
||
self._parse()
|
||
|
||
def _parse(self):
|
||
with open(self.ioc_path, encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
if '=' in line:
|
||
key, value = line.split('=', 1)
|
||
self.config[key.strip()] = value.strip()
|
||
|
||
def is_freertos_enabled(self):
|
||
ip_keys = [k for k in self.config if k.startswith('Mcu.IP')]
|
||
for k in ip_keys:
|
||
if self.config[k] == 'FREERTOS':
|
||
return True
|
||
for k in self.config:
|
||
if k.startswith('FREERTOS.'):
|
||
return True
|
||
return False
|
||
|
||
class DataInterface(QWidget):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent=parent)
|
||
self.setObjectName("dataInterface")
|
||
|
||
# 属性初始化
|
||
self.project_path = ""
|
||
self.project_name = ""
|
||
self.ioc_file = ""
|
||
self.freertos_enabled = False # 新增属性
|
||
|
||
# 主布局
|
||
self.stacked_layout = QStackedLayout(self)
|
||
self.setLayout(self.stacked_layout)
|
||
|
||
# --- 页面1：工程路径选择 ---
|
||
self.select_widget = QWidget()
|
||
outer_layout = QVBoxLayout(self.select_widget)
|
||
outer_layout.setContentsMargins(0, 0, 0, 0)
|
||
outer_layout.addStretch()
|
||
|
||
# 直接用布局和控件，无卡片
|
||
content_layout = QVBoxLayout()
|
||
content_layout.setSpacing(28)
|
||
content_layout.setContentsMargins(48, 48, 48, 48)
|
||
|
||
# 主标题
|
||
title = TitleLabel("MRobot 代码生成")
|
||
title.setAlignment(Qt.AlignCenter)
|
||
content_layout.addWidget(title)
|
||
|
||
# 副标题
|
||
subtitle = BodyLabel("请选择您的由CUBEMX生成的工程路径（.ico所在的目录），然后开启代码之旅！")
|
||
subtitle.setAlignment(Qt.AlignCenter)
|
||
content_layout.addWidget(subtitle)
|
||
|
||
# 简要说明
|
||
desc = BodyLabel("支持自动配置和生成任务，自主选择模块代码倒入，自动识别cubemx配置！")
|
||
desc.setAlignment(Qt.AlignCenter)
|
||
content_layout.addWidget(desc)
|
||
|
||
content_layout.addSpacing(18)
|
||
|
||
# 选择项目路径按钮
|
||
self.choose_btn = PushButton(FluentIcon.FOLDER, "选择项目路径")
|
||
self.choose_btn.setFixedWidth(200)
|
||
self.choose_btn.clicked.connect(self.choose_project_folder)
|
||
content_layout.addWidget(self.choose_btn, alignment=Qt.AlignmentFlag.AlignCenter)
|
||
|
||
# 更新代码库按钮
|
||
self.update_template_btn = PushButton(FluentIcon.SYNC, "更新代码库")
|
||
self.update_template_btn.setFixedWidth(200)
|
||
self.update_template_btn.clicked.connect(self.update_user_template)
|
||
content_layout.addWidget(self.update_template_btn, alignment=Qt.AlignmentFlag.AlignCenter)
|
||
|
||
content_layout.addSpacing(10)
|
||
content_layout.addStretch()
|
||
|
||
|
||
outer_layout.addLayout(content_layout)
|
||
outer_layout.addStretch()
|
||
|
||
self.stacked_layout.addWidget(self.select_widget)
|
||
|
||
|
||
# --- 页面2：主配置页面 ---
|
||
self.config_widget = QWidget()
|
||
main_layout = QVBoxLayout(self.config_widget)
|
||
main_layout.setContentsMargins(32, 32, 32, 32)
|
||
main_layout.setSpacing(18)
|
||
|
||
# 顶部项目信息
|
||
info_layout = QHBoxLayout()
|
||
self.back_btn = PushButton(FluentIcon.SKIP_BACK, "返回")
|
||
self.back_btn.setFixedWidth(90)
|
||
self.back_btn.clicked.connect(self.back_to_select)
|
||
info_layout.addWidget(self.back_btn) # 返回按钮放最左
|
||
self.project_name_label = StrongBodyLabel()
|
||
self.project_path_label = BodyLabel()
|
||
self.ioc_file_label = BodyLabel()
|
||
self.freertos_label = BodyLabel()
|
||
info_layout.addWidget(self.project_name_label)
|
||
info_layout.addWidget(self.project_path_label)
|
||
info_layout.addWidget(self.ioc_file_label)
|
||
info_layout.addWidget(self.freertos_label)
|
||
info_layout.addStretch()
|
||
main_layout.addLayout(info_layout)
|
||
main_layout.addWidget(HorizontalSeparator())
|
||
|
||
# ======= 新增：左右分栏 =======
|
||
content_hbox = QHBoxLayout()
|
||
content_hbox.setSpacing(24)
|
||
|
||
# 左侧：文件树
|
||
left_vbox = QVBoxLayout()
|
||
left_vbox.addWidget(SubtitleLabel("用户代码模块选择"))
|
||
left_vbox.addWidget(HorizontalSeparator())
|
||
self.file_tree = TreeWidget()
|
||
self.file_tree.setHeaderLabels(["模块名"])
|
||
self.file_tree.setSelectionMode(self.file_tree.ExtendedSelection)
|
||
self.file_tree.header().setSectionResizeMode(0, QHeaderView.Stretch)
|
||
self.file_tree.setCheckedColor("#0078d4", "#2d7d9a")
|
||
self.file_tree.setBorderRadius(8)
|
||
self.file_tree.setBorderVisible(True)
|
||
left_vbox.addWidget(self.file_tree, stretch=1)
|
||
content_hbox.addLayout(left_vbox, 2)
|
||
|
||
# 右侧：操作按钮和说明
|
||
right_vbox = QVBoxLayout()
|
||
right_vbox.setSpacing(18)
|
||
right_vbox.addWidget(SubtitleLabel("操作区"))
|
||
right_vbox.addWidget(HorizontalSeparator())
|
||
|
||
# 操作按钮分组
|
||
btn_group = QVBoxLayout()
|
||
# 自动环境配置按钮
|
||
self.env_btn = PushButton("自动环境配置")
|
||
self.env_btn.setFixedWidth(200)
|
||
self.env_btn.setToolTip("自动检测并配置常用开发环境（功能开发中）")
|
||
self.env_btn.clicked.connect(self.auto_env_config)
|
||
btn_group.addWidget(self.env_btn)
|
||
# FreeRTOS相关按钮
|
||
self.freertos_task_btn = PushButton("自动生成FreeRTOS任务")
|
||
self.freertos_task_btn.setFixedWidth(200)
|
||
self.freertos_task_btn.setToolTip("自动在 freertos.c 中插入任务创建代码")
|
||
self.freertos_task_btn.clicked.connect(self.on_freertos_task_btn_clicked)
|
||
btn_group.addWidget(self.freertos_task_btn)
|
||
self.task_code_btn = PushButton("配置并生成任务代码")
|
||
self.task_code_btn.setFixedWidth(200)
|
||
self.task_code_btn.setToolTip("配置任务参数并一键生成任务代码文件")
|
||
self.task_code_btn.clicked.connect(self.on_task_code_btn_clicked)
|
||
btn_group.addWidget(self.task_code_btn)
|
||
self.generate_btn = PushButton(FluentIcon.CODE, "生成代码")
|
||
self.generate_btn.setFixedWidth(200)
|
||
self.generate_btn.setToolTip("将选中的用户模块代码复制到工程 User 目录")
|
||
self.generate_btn.clicked.connect(self.generate_code)
|
||
btn_group.addWidget(self.generate_btn)
|
||
btn_group.addSpacing(10)
|
||
right_vbox.addLayout(btn_group)
|
||
right_vbox.addStretch()
|
||
|
||
content_hbox.addLayout(right_vbox, 1)
|
||
main_layout.addLayout(content_hbox, stretch=1)
|
||
self.stacked_layout.addWidget(self.config_widget)
|
||
self.file_tree.itemChanged.connect(self.on_tree_item_changed)
|
||
|
||
def auto_env_config(self):
|
||
InfoBar.info(
|
||
title="敬请期待",
|
||
content="自动环境配置功能暂未实现，等待后续更新。",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
|
||
def choose_project_folder(self):
|
||
folder = QFileDialog.getExistingDirectory(self, "请选择代码项目文件夹")
|
||
if not folder:
|
||
return
|
||
ioc_files = [f for f in os.listdir(folder) if f.endswith('.ioc')]
|
||
if not ioc_files:
|
||
InfoBar.warning(
|
||
title="提示",
|
||
content="未找到.ioc文件，请确认项目文件夹。",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
return
|
||
self.project_path = folder
|
||
self.project_name = os.path.basename(folder)
|
||
self.ioc_file = os.path.join(folder, ioc_files[0])
|
||
self.show_config_page()
|
||
|
||
def show_config_page(self):
|
||
# 更新项目信息
|
||
self.project_name_label.setText(f"项目名称: {self.project_name}")
|
||
self.project_path_label.setText(f"项目路径: {self.project_path}")
|
||
# self.ioc_file_label.setText(f"IOC 文件: {self.ioc_file}")
|
||
try:
|
||
ioc = IocConfig(self.ioc_file)
|
||
self.freertos_enabled = ioc.is_freertos_enabled() # 记录状态
|
||
freertos_status = "已启用" if self.freertos_enabled else "未启用"
|
||
self.freertos_label.setText(f"FreeRTOS: {freertos_status}")
|
||
# self.freertos_task_btn.setEnabled(self.freertos_enabled)
|
||
except Exception as e:
|
||
self.freertos_label.setText(f"IOC解析失败: {e}")
|
||
self.freertos_task_btn.hide()
|
||
self.freertos_enabled = False
|
||
self.show_user_code_files()
|
||
self.stacked_layout.setCurrentWidget(self.config_widget)
|
||
|
||
def on_freertos_task_btn_clicked(self):
|
||
if not self.freertos_enabled:
|
||
InfoBar.warning(
|
||
title="未开启 FreeRTOS",
|
||
content="请先在 CubeMX 中开启 FreeRTOS！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
return
|
||
self.generate_freertos_task()
|
||
|
||
def on_task_code_btn_clicked(self):
|
||
if not self.freertos_enabled:
|
||
InfoBar.warning(
|
||
title="未开启 FreeRTOS",
|
||
content="请先在 CubeMX 中开启 FreeRTOS！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
return
|
||
self.open_task_config_dialog()
|
||
|
||
def back_to_select(self):
|
||
self.stacked_layout.setCurrentWidget(self.select_widget)
|
||
|
||
def update_user_template(self):
|
||
url = "http://gitea.qutrobot.top/robofish/MRobot/archive/User_code.zip"
|
||
local_dir = "User_code"
|
||
try:
|
||
resp = requests.get(url, timeout=30)
|
||
resp.raise_for_status()
|
||
z = zipfile.ZipFile(io.BytesIO(resp.content))
|
||
if os.path.exists(local_dir):
|
||
shutil.rmtree(local_dir)
|
||
for member in z.namelist():
|
||
rel_path = os.path.relpath(member, z.namelist()[0])
|
||
if rel_path == ".":
|
||
continue
|
||
target_path = os.path.join(local_dir, rel_path)
|
||
if member.endswith('/'):
|
||
os.makedirs(target_path, exist_ok=True)
|
||
else:
|
||
os.makedirs(os.path.dirname(target_path), exist_ok=True)
|
||
with open(target_path, "wb") as f:
|
||
f.write(z.read(member))
|
||
InfoBar.success(
|
||
title="更新成功",
|
||
content="用户模板已更新到最新版本！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
except Exception as e:
|
||
InfoBar.error(
|
||
title="更新失败",
|
||
content=f"用户模板更新失败: {e}",
|
||
parent=self,
|
||
duration=3000
|
||
)
|
||
|
||
def show_user_code_files(self):
|
||
self.file_tree.clear()
|
||
base_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../User_code")
|
||
user_dir = os.path.join(self.project_path, "User")
|
||
sub_dirs = ["bsp", "component", "device", "module"]
|
||
|
||
# 读取所有 describe.csv 和 dependencies.csv
|
||
describe_map = {}
|
||
dependencies_map = {}
|
||
for sub in sub_dirs:
|
||
dir_path = os.path.join(base_dir, sub)
|
||
if not os.path.isdir(dir_path):
|
||
continue
|
||
# describe
|
||
desc_path = os.path.join(dir_path, "describe.csv")
|
||
if os.path.exists(desc_path):
|
||
with open(desc_path, encoding="utf-8") as f:
|
||
for line in f:
|
||
if "," in line:
|
||
k, v = line.strip().split(",", 1)
|
||
describe_map[f"{sub}/{k.strip()}"] = v.strip()
|
||
# dependencies
|
||
dep_path = os.path.join(dir_path, "dependencies.csv")
|
||
if os.path.exists(dep_path):
|
||
with open(dep_path, encoding="utf-8") as f:
|
||
for line in f:
|
||
if "," in line:
|
||
a, b = line.strip().split(",", 1)
|
||
dependencies_map.setdefault(f"{sub}/{a.strip()}", []).append(b.strip())
|
||
|
||
self._describe_map = describe_map
|
||
self._dependencies_map = dependencies_map
|
||
|
||
self.file_tree.setHeaderLabels(["模块名", "描述"])
|
||
self.file_tree.setSelectionMode(self.file_tree.ExtendedSelection)
|
||
self.file_tree.header().setSectionResizeMode(0, QHeaderView.ResizeToContents)
|
||
self.file_tree.header().setSectionResizeMode(1, QHeaderView.Stretch) # 描述列自适应
|
||
self.file_tree.setCheckedColor("#0078d4", "#2d7d9a")
|
||
self.file_tree.setBorderRadius(8)
|
||
self.file_tree.setBorderVisible(True)
|
||
|
||
for sub in sub_dirs:
|
||
dir_path = os.path.join(base_dir, sub)
|
||
if not os.path.isdir(dir_path):
|
||
continue
|
||
group_item = TreeItem([sub, ""])
|
||
self.file_tree.addTopLevelItem(group_item)
|
||
has_file = False
|
||
for root, _, files in os.walk(dir_path):
|
||
rel_root = os.path.relpath(root, base_dir)
|
||
for f in sorted(files):
|
||
if f.endswith(".c"):
|
||
mod_name = os.path.splitext(f)[0]
|
||
rel_c = os.path.join(rel_root, f)
|
||
key = f"{rel_root}/{mod_name}".replace("\\", "/")
|
||
desc = describe_map.get(key, "")
|
||
file_item = TreeItem([mod_name, desc])
|
||
file_item.setFlags(file_item.flags() | Qt.ItemIsUserCheckable)
|
||
file_item.setData(0, Qt.UserRole, rel_c)
|
||
file_item.setData(0, Qt.UserRole + 1, key) # 存模块key
|
||
file_item.setToolTip(1, desc)
|
||
file_item.setTextAlignment(1, Qt.AlignLeft | Qt.AlignVCenter)
|
||
group_item.addChild(file_item)
|
||
dst_c = os.path.join(user_dir, rel_c)
|
||
if os.path.exists(dst_c):
|
||
file_item.setCheckState(0, Qt.Unchecked)
|
||
file_item.setText(0, f"{mod_name}（已存在）")
|
||
file_item.setForeground(0, Qt.gray)
|
||
else:
|
||
file_item.setCheckState(0, Qt.Unchecked)
|
||
group_item.addChild(file_item)
|
||
has_file = True
|
||
if not has_file:
|
||
empty_item = TreeItem(["（无 .c 文件）", ""])
|
||
group_item.addChild(empty_item)
|
||
self.file_tree.expandAll()
|
||
|
||
# 勾选依赖自动勾选
|
||
def on_tree_item_changed(self, item, column):
|
||
if column != 0:
|
||
return
|
||
if item.childCount() > 0:
|
||
return # 只处理叶子
|
||
if item.checkState(0) == Qt.Checked:
|
||
key = item.data(0, Qt.UserRole + 1)
|
||
deps = self._dependencies_map.get(key, [])
|
||
if deps:
|
||
checked = []
|
||
root = self.file_tree.invisibleRootItem()
|
||
for i in range(root.childCount()):
|
||
group = root.child(i)
|
||
for j in range(group.childCount()):
|
||
child = group.child(j)
|
||
ckey = child.data(0, Qt.UserRole + 1)
|
||
if ckey in deps and child.checkState(0) != Qt.Checked:
|
||
child.setCheckState(0, Qt.Checked)
|
||
checked.append(ckey)
|
||
if checked:
|
||
descs = [self._describe_map.get(dep, dep) for dep in checked]
|
||
InfoBar.info(
|
||
title="依赖自动勾选",
|
||
content="已自动勾选依赖模块: " + "，".join(descs),
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
|
||
|
||
def get_checked_files(self):
|
||
files = []
|
||
def _traverse(item):
|
||
for i in range(item.childCount()):
|
||
child = item.child(i)
|
||
if child.childCount() == 0 and child.checkState(0) == Qt.Checked:
|
||
files.append(child.data(0, Qt.UserRole))
|
||
_traverse(child)
|
||
root = self.file_tree.invisibleRootItem()
|
||
for i in range(root.childCount()):
|
||
_traverse(root.child(i))
|
||
return files
|
||
|
||
def generate_code(self):
|
||
import shutil
|
||
base_dir = "User_code"
|
||
user_dir = os.path.join(self.project_path, "User")
|
||
copied = []
|
||
files = self.get_checked_files()
|
||
skipped = []
|
||
for rel_c in files:
|
||
rel_h = rel_c[:-2] + ".h"
|
||
src_c = os.path.join(base_dir, rel_c)
|
||
src_h = os.path.join(base_dir, rel_h)
|
||
dst_c = os.path.join(user_dir, rel_c)
|
||
dst_h = os.path.join(user_dir, rel_h)
|
||
# 如果目标文件已存在则跳过
|
||
if os.path.exists(dst_c):
|
||
skipped.append(dst_c)
|
||
else:
|
||
os.makedirs(os.path.dirname(dst_c), exist_ok=True)
|
||
shutil.copy2(src_c, dst_c)
|
||
copied.append(dst_c)
|
||
if os.path.exists(src_h):
|
||
if os.path.exists(dst_h):
|
||
skipped.append(dst_h)
|
||
else:
|
||
os.makedirs(os.path.dirname(dst_h), exist_ok=True)
|
||
shutil.copy2(src_h, dst_h)
|
||
copied.append(dst_h)
|
||
msg = f"已拷贝 {len(copied)} 个文件到 User 目录"
|
||
if skipped:
|
||
msg += f"\n{len(skipped)} 个文件已存在，未覆盖"
|
||
InfoBar.success(
|
||
title="生成完成",
|
||
content=msg,
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
# 生成后刷新文件树，更新标记
|
||
self.show_user_code_files()
|
||
|
||
def generate_freertos_task(self):
|
||
import re
|
||
freertos_path = os.path.join(self.project_path, "Core", "Src", "freertos.c")
|
||
if not os.path.exists(freertos_path):
|
||
InfoBar.error(
|
||
title="未找到 freertos.c",
|
||
content="未找到 Core/Src/freertos.c 文件，请确认工程路径。",
|
||
parent=self,
|
||
duration=2500
|
||
)
|
||
return
|
||
with open(freertos_path, "r", encoding="utf-8") as f:
|
||
code = f.read()
|
||
|
||
changed = False
|
||
error_msgs = []
|
||
|
||
# 1. 添加 #include "task/user_task.h"
|
||
include_line = '#include "task/user_task.h"'
|
||
if include_line not in code:
|
||
# 只插入到 USER CODE BEGIN Includes 区域
|
||
include_pattern = r'(\/\* *USER CODE BEGIN Includes *\*\/\s*)'
|
||
if re.search(include_pattern, code):
|
||
code = re.sub(
|
||
include_pattern,
|
||
r'\1' + include_line + '\n',
|
||
code
|
||
)
|
||
changed = True
|
||
else:
|
||
error_msgs.append("未找到 /* USER CODE BEGIN Includes */ 区域，无法插入 include。")
|
||
|
||
# 2. 在 /* USER CODE BEGIN RTOS_THREADS */ 区域添加 osThreadNew(Task_Init, NULL, &attr_init);
|
||
rtos_threads_pattern = r'(\/\* *USER CODE BEGIN RTOS_THREADS *\*\/\s*)(.*?)(\/\* *USER CODE END RTOS_THREADS *\*\/)'
|
||
match = re.search(rtos_threads_pattern, code, re.DOTALL)
|
||
task_line = ' osThreadNew(Task_Init, NULL, &attr_init); // 创建初始化任务\n'
|
||
if match:
|
||
threads_code = match.group(2)
|
||
if 'Task_Init' not in threads_code:
|
||
# 保留原有内容，追加新行
|
||
new_threads_code = match.group(1) + threads_code + task_line + match.group(3)
|
||
code = code[:match.start()] + new_threads_code + code[match.end():]
|
||
changed = True
|
||
else:
|
||
error_msgs.append("未找到 /* USER CODE BEGIN RTOS_THREADS */ 区域，无法插入任务创建代码。")
|
||
|
||
# 3. 清空 StartDefaultTask 的 USER CODE 区域，只保留 osThreadTerminate
|
||
sdt_pattern = r'(\/\* *USER CODE BEGIN StartDefaultTask *\*\/\s*)(.*?)(\/\* *USER CODE END StartDefaultTask *\*\/)'
|
||
match = re.search(sdt_pattern, code, re.DOTALL)
|
||
if match:
|
||
if 'osThreadTerminate(osThreadGetId());' not in match.group(2):
|
||
new_sdt_code = match.group(1) + ' osThreadTerminate(osThreadGetId());\n' + match.group(3)
|
||
code = code[:match.start()] + new_sdt_code + code[match.end():]
|
||
changed = True
|
||
else:
|
||
error_msgs.append("未找到 /* USER CODE BEGIN StartDefaultTask */ 区域，无法插入终止代码。")
|
||
|
||
if changed:
|
||
with open(freertos_path, "w", encoding="utf-8") as f:
|
||
f.write(code)
|
||
InfoBar.success(
|
||
title="生成成功",
|
||
content="FreeRTOS任务代码已自动生成！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
elif error_msgs:
|
||
InfoBar.error(
|
||
title="生成失败",
|
||
content="\n".join(error_msgs),
|
||
parent=self,
|
||
duration=3000
|
||
)
|
||
else:
|
||
InfoBar.info(
|
||
title="无需修改",
|
||
content="FreeRTOS任务相关代码已存在，无需重复生成。",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
|
||
def open_task_config_dialog(self):
|
||
from PyQt5.QtWidgets import QDialog, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QSpinBox, QPushButton, QTableWidget, QTableWidgetItem, QHeaderView, QCheckBox
|
||
import yaml
|
||
import os
|
||
|
||
class TaskConfigDialog(QDialog):
|
||
def __init__(self, parent=None, config_path=None):
|
||
super().__init__(parent)
|
||
self.setWindowTitle("任务配置")
|
||
self.resize(900, 420)
|
||
layout = QVBoxLayout(self)
|
||
self.table = QTableWidget(0, 6)
|
||
self.table.setHorizontalHeaderLabels(["任务名称", "运行频率", "初始化延迟", "堆栈大小", "任务描述", "频率控制"])
|
||
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
|
||
self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
|
||
self.table.horizontalHeader().setSectionResizeMode(2, QHeaderView.Stretch)
|
||
self.table.horizontalHeader().setSectionResizeMode(3, QHeaderView.Stretch)
|
||
self.table.horizontalHeader().setSectionResizeMode(4, QHeaderView.Stretch)
|
||
self.table.horizontalHeader().setSectionResizeMode(5, QHeaderView.ResizeToContents)
|
||
self.table.setColumnWidth(4, 320) # 任务描述更宽
|
||
layout.addWidget(self.table)
|
||
btn_layout = QHBoxLayout()
|
||
add_btn = QPushButton("添加任务")
|
||
del_btn = QPushButton("删除选中")
|
||
ok_btn = QPushButton("生成")
|
||
cancel_btn = QPushButton("取消")
|
||
btn_layout.addWidget(add_btn)
|
||
btn_layout.addWidget(del_btn)
|
||
btn_layout.addStretch()
|
||
btn_layout.addWidget(ok_btn)
|
||
btn_layout.addWidget(cancel_btn)
|
||
layout.addLayout(btn_layout)
|
||
add_btn.clicked.connect(self.add_row)
|
||
del_btn.clicked.connect(self.del_row)
|
||
ok_btn.clicked.connect(self.accept)
|
||
cancel_btn.clicked.connect(self.reject)
|
||
|
||
|
||
# 自动读取配置文件
|
||
if config_path and os.path.exists(config_path):
|
||
try:
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
tasks = yaml.safe_load(f)
|
||
if tasks:
|
||
for t in tasks:
|
||
row = self.table.rowCount()
|
||
self.table.insertRow(row)
|
||
for col, key in enumerate(["name", "frequency", "delay", "stack", "description"]):
|
||
item = QTableWidgetItem(str(t.get(key, "")))
|
||
item.setTextAlignment(Qt.AlignCenter)
|
||
self.table.setItem(row, col, item)
|
||
# 新增频率控制复选框
|
||
freq_ctrl = QCheckBox()
|
||
freq_ctrl.setChecked(t.get("freq_control", True))
|
||
self.table.setCellWidget(row, 5, freq_ctrl)
|
||
except Exception as e:
|
||
pass # 配置文件损坏时忽略
|
||
|
||
def add_row(self):
|
||
row = self.table.rowCount()
|
||
self.table.insertRow(row)
|
||
default_values = [
|
||
f"Task{row+1}", "500", "0", "256", "不要偷懒，请写清楚每个任务的作用！（如果你看到任务上面是这句话，说明作者是个懒蛋）"
|
||
]
|
||
for col, val in enumerate(default_values):
|
||
item = QTableWidgetItem(val)
|
||
item.setTextAlignment(Qt.AlignCenter)
|
||
self.table.setItem(row, col, item)
|
||
freq_ctrl = QCheckBox()
|
||
freq_ctrl.setChecked(True)
|
||
self.table.setCellWidget(row, 5, freq_ctrl)
|
||
|
||
def del_row(self):
|
||
rows = set([i.row() for i in self.table.selectedItems()])
|
||
for r in sorted(rows, reverse=True):
|
||
self.table.removeRow(r)
|
||
|
||
def get_tasks(self):
|
||
tasks = []
|
||
for row in range(self.table.rowCount()):
|
||
name = self.table.item(row, 0).text().strip()
|
||
freq = self.table.item(row, 1).text()
|
||
delay = int(self.table.item(row, 2).text())
|
||
stack = int(self.table.item(row, 3).text())
|
||
desc = self.table.item(row, 4).text().strip()
|
||
freq_ctrl = self.table.cellWidget(row, 5).isChecked()
|
||
# 校验 stack 必须为 128*2^n
|
||
if stack < 128 or (stack & (stack - 1)) != 0 or stack % 128 != 0:
|
||
raise ValueError(f"第{row+1}行任务“{name}”的堆栈大小必须为128、256、512、1024等（128*2^n）")
|
||
task = {
|
||
"name": name,
|
||
"function": f"Task_{name}",
|
||
"delay": delay,
|
||
"stack": stack,
|
||
"description": desc,
|
||
"freq_control": freq_ctrl
|
||
}
|
||
if freq_ctrl:
|
||
task["frequency"] = int(freq)
|
||
tasks.append(task)
|
||
return tasks
|
||
|
||
|
||
|
||
|
||
|
||
config_path = os.path.join(self.project_path, "User", "task", "config.yaml")
|
||
dlg = TaskConfigDialog(self, config_path=config_path)
|
||
if dlg.exec() == QDialog.Accepted:
|
||
try:
|
||
tasks = dlg.get_tasks()
|
||
except Exception as e:
|
||
InfoBar.error(
|
||
title="参数错误",
|
||
content=str(e),
|
||
parent=self,
|
||
duration=3000
|
||
)
|
||
return
|
||
if not tasks:
|
||
InfoBar.warning(
|
||
title="未配置任务",
|
||
content="请至少添加一个任务！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
return
|
||
try:
|
||
self.generate_task_code(tasks)
|
||
InfoBar.success(
|
||
title="生成成功",
|
||
content="任务代码已生成到 User/task 目录！",
|
||
parent=self,
|
||
duration=2000
|
||
)
|
||
except Exception as e:
|
||
InfoBar.error(
|
||
title="生成失败",
|
||
content=f"任务代码生成失败: {e}",
|
||
parent=self,
|
||
duration=3000
|
||
)
|
||
|
||
def preserve_user_region(self, new_code, old_code, region_name):
|
||
"""
|
||
替换 new_code 中 region_name 区域为 old_code 中的内容（如果有）
|
||
region_name: 如 'USER INCLUDE'
|
||
"""
|
||
pattern = re.compile(
|
||
rf"/\*\s*{region_name}\s*BEGIN\s*\*/(.*?)/\*\s*{region_name}\s*END\s*\*/",
|
||
re.DOTALL
|
||
)
|
||
old_match = pattern.search(old_code or "")
|
||
if not old_match:
|
||
return new_code # 旧文件没有该区域，直接返回新代码
|
||
|
||
old_content = old_match.group(1)
|
||
def repl(m):
|
||
return m.group(0).replace(m.group(1), old_content)
|
||
# 替换新代码中的该区域
|
||
return pattern.sub(repl, new_code, count=1)
|
||
|
||
def generate_task_code(self, task_list):
|
||
|
||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||
template_dir = os.path.join(base_dir, "User_code", "task")
|
||
output_dir = os.path.join(self.project_path, "User", "task")
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
user_task_h_tpl = os.path.join(template_dir, "user_task.h.template")
|
||
user_task_c_tpl = os.path.join(template_dir, "user_task.c.template")
|
||
init_c_tpl = os.path.join(template_dir, "init.c.template")
|
||
task_c_tpl = os.path.join(template_dir, "task.c.template")
|
||
|
||
freq_tasks = [t for t in task_list if t.get("freq_control", True)]
|
||
|
||
def render_template(path, context):
|
||
with open(path, encoding="utf-8") as f:
|
||
tpl = Template(f.read())
|
||
return tpl.render(**context)
|
||
|
||
context_h = {
|
||
"thread_definitions": "\n".join([f" osThreadId_t {t['name']};" for t in task_list]),
|
||
"freq_definitions": "\n".join([f" float {t['name']};" for t in freq_tasks]),
|
||
"stack_definitions": "\n".join([f" UBaseType_t {t['name']};" for t in task_list]),
|
||
"last_up_time_definitions": "\n".join([f" float {t['name']};" for t in freq_tasks]),
|
||
"task_frequency_definitions": "\n".join([f"#define {t['name'].upper()}_FREQ ({t['frequency']})" for t in freq_tasks]),
|
||
"task_init_delay_definitions": "\n".join([f"#define {t['name'].upper()}_INIT_DELAY ({t['delay']})" for t in task_list]),
|
||
"task_attr_declarations": "\n".join([f"extern const osThreadAttr_t attr_{t['name']};" for t in task_list]),
|
||
"task_function_declarations": "\n".join([f"void {t['function']}(void *argument);" for t in task_list]),
|
||
}
|
||
|
||
# ----------- 生成 user_task.h -----------
|
||
user_task_h_path = os.path.join(output_dir, "user_task.h")
|
||
new_user_task_h = render_template(user_task_h_tpl, context_h)
|
||
|
||
if os.path.exists(user_task_h_path):
|
||
with open(user_task_h_path, "r", encoding="utf-8") as f:
|
||
old_code = f.read()
|
||
for region in ["USER INCLUDE", "USER MESSAGE", "USER CONFIG"]:
|
||
pattern = re.compile(
|
||
rf"/\*\s*{region}\s*BEGIN\s*\*/(.*?)/\*\s*{region}\s*END\s*\*/",
|
||
re.DOTALL
|
||
)
|
||
old_match = pattern.search(old_code)
|
||
if old_match and old_match.group(1).strip():
|
||
new_user_task_h = self.preserve_user_region(
|
||
new_user_task_h, old_code, region
|
||
)
|
||
with open(user_task_h_path, "w", encoding="utf-8") as f:
|
||
f.write(new_user_task_h)
|
||
|
||
# ----------- 生成 user_task.c -----------
|
||
context_c = {
|
||
"task_attr_definitions": "\n".join([
|
||
f"const osThreadAttr_t attr_{t['name']} = {{\n"
|
||
f" .name = \"{t['name']}\",\n"
|
||
f" .priority = osPriorityNormal,\n"
|
||
f" .stack_size = {t['stack']} * 4,\n"
|
||
f"}};"
|
||
for t in task_list
|
||
])
|
||
}
|
||
user_task_c = render_template(user_task_c_tpl, context_c)
|
||
with open(os.path.join(output_dir, "user_task.c"), "w", encoding="utf-8") as f:
|
||
f.write(user_task_c)
|
||
|
||
# ----------- 生成 init.c -----------
|
||
thread_creation_code = "\n".join([
|
||
f" task_runtime.thread.{t['name']} = osThreadNew({t['function']}, NULL, &attr_{t['name']});"
|
||
for t in task_list
|
||
])
|
||
context_init = {
|
||
"thread_creation_code": thread_creation_code,
|
||
}
|
||
init_c = render_template(init_c_tpl, context_init)
|
||
init_c_path = os.path.join(output_dir, "init.c")
|
||
if os.path.exists(init_c_path):
|
||
with open(init_c_path, "r", encoding="utf-8") as f:
|
||
old_code = f.read()
|
||
for region in ["USER INCLUDE", "USER CODE", "USER CODE INIT"]:
|
||
init_c = self.preserve_user_region(
|
||
init_c, old_code, region
|
||
)
|
||
with open(init_c_path, "w", encoding="utf-8") as f:
|
||
f.write(init_c)
|
||
|
||
# ----------- 生成 task.c -----------
|
||
for t in task_list:
|
||
desc = t.get("description", "")
|
||
desc_wrapped = "\n ".join(textwrap.wrap(desc, 20))
|
||
context_task = {
|
||
"task_name": t["name"],
|
||
"task_function": t["function"],
|
||
"task_frequency": f"{t['name'].upper()}_FREQ" if t.get("freq_control", True) else None,
|
||
"task_delay": f"{t['name'].upper()}_INIT_DELAY",
|
||
"task_description": desc_wrapped,
|
||
"freq_control": t.get("freq_control", True)
|
||
}
|
||
with open(task_c_tpl, encoding="utf-8") as f:
|
||
tpl = Template(f.read())
|
||
code = tpl.render(**context_task)
|
||
task_c_path = os.path.join(output_dir, f"{t['name']}.c")
|
||
if os.path.exists(task_c_path):
|
||
with open(task_c_path, "r", encoding="utf-8") as f:
|
||
old_code = f.read()
|
||
for region in ["USER INCLUDE", "USER STRUCT", "USER CODE", "USER CODE INIT"]:
|
||
code = self.preserve_user_region(
|
||
code, old_code, region
|
||
)
|
||
with open(task_c_path, "w", encoding="utf-8") as f:
|
||
f.write(code)
|
||
|
||
# ----------- 保存任务配置到 config.yaml -----------
|
||
config_yaml_path = os.path.join(output_dir, "config.yaml")
|
||
with open(config_yaml_path, "w", encoding="utf-8") as f:
|
||
yaml.safe_dump(task_list, f, allow_unicode=True)
|
||
|