mirror of
https://github.com/goldenfishs/MRobot.git
synced 2026-03-31 21:07:14 +08:00
重构架构
This commit is contained in:
810
app/data_interface.py
Normal file
810
app/data_interface.py
Normal file
@@ -0,0 +1,810 @@
|
||||
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)
|
||||
title.setStyleSheet("font-size: 36px; font-weight: bold; color: #2d7d9a;")
|
||||
content_layout.addWidget(title)
|
||||
|
||||
# 副标题
|
||||
subtitle = BodyLabel("请选择您的由CUBEMX生成的工程路径（.ico所在的目录），然后开启代码之旅！")
|
||||
subtitle.setAlignment(Qt.AlignCenter)
|
||||
subtitle.setStyleSheet("font-size: 16px; color: #4a6fa5;")
|
||||
content_layout.addWidget(subtitle)
|
||||
|
||||
# 简要说明
|
||||
desc = BodyLabel("支持自动配置和生成任务，自主选择模块代码倒入，自动识别cubemx配置！")
|
||||
desc.setAlignment(Qt.AlignCenter)
|
||||
desc.setStyleSheet("font-size: 14px; color: #6b7b8c;")
|
||||
content_layout.addWidget(desc)
|
||||
|
||||
content_layout.addSpacing(18)
|
||||
|
||||
# 选择项目路径按钮
|
||||
self.choose_btn = PushButton(FluentIcon.FOLDER, "选择项目路径")
|
||||
self.choose_btn.setFixedWidth(200)
|
||||
self.choose_btn.setStyleSheet("font-size: 17px;")
|
||||
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.setStyleSheet("font-size: 17px;")
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user