这段代码构建了一个比较完整的 MVC (Model-View-Controller) 架构的桌面应用程序,涉及数据库操作、UI展示和后台仿真任务。
整体来看,代码结构清晰,分层合理,已经具备了生产级应用的雏形。但是,在 数据流的一致性、ORM的使用细节、MVC解耦 以及 线程安全 方面,存在一些设计隐患和改进空间。
以下是详细的审查报告:
1. 核心架构问题:数据“双重来源”风险
这是当前设计中最大的隐患。
建议:
确保每次加载新产品时 (display_design_target),不仅要填入 UI,还要同步重置/初始化 DataService 的状态,确保它只包含当前产品的参数。
2. Peewee ORM 的使用细节 (Model 层)
A. product.detail 的访问方式不一致 (易报错)
在 model.py 中:你使用了 product.detail.get() (这是获取单个对象)。
在 bldc_page.py 中:你使用了 product.detail[0] (这是把 backref 当作列表索引)。
detail_object = product.detail[0] if product.detail else None
分析:Peewee 的 backref 默认返回一个 SelectQuery。
建议:统一使用一种安全的方法。
detail_object = product.detail.first()
B. find_product_by_id 中的性能隐患 (副作用)
def find_product_by_id(product_id: int):
product_dict = model_to_dict(product, backrefs=True, recurse=True)
for k, v in product_dict.items():
print(f"{k}: {v}")
return product
C. 数据库连接管理
问题:initialize_database 和 close_connection 是全局函数。在 controller.py 的 shutdown 中调用关闭。
风险:如果在多线程环境(例如 MaxwellWorker 线程)中访问数据库,SQLite 需要特别小心。虽然 Peewee 的 SqliteDatabase 默认有一定的线程安全处理,但建议在 Worker 线程中不要复用主线程的连接实例,或者确保 SQLite 编译时支持多线程且连接模式设置为 WAL。
建议:目前你的 Worker 主要是跑 Maxwell 脚本,如果 Worker 结束后要写回数据库(更新结果),请确保使用 db.connection_context() 或在任务开始时建立连接,结束时关闭。
3. MVC 耦合度问题 (Controller 层)
A. Controller 对 View 的侵入性太强
代码:
self._view.customer_page.product_combo_box.currentIndexChanged...
self._view.customer_page.get_selected_product_id()
self._view.customer_page.load_data_to_form(product)
问题:Controller 深入到了 View 的内部组件(product_combo_box)。如果 View 改版了(比如把下拉框换成了列表),Controller 必须重写。
建议:遵循 “迪米特法则” (Law of Demeter)。
View 应该暴露出信号和高级方法。
View 定义:signal_product_selected = pyqtSignal(int)
Controller 连接:view.customer_page.signal_product_selected.connect(self._on_product_selected)
这样 Controller 不需要知道 View 用的是 ComboBox 还是 ListView。
B. 异常处理过于宽泛
代码:except Exception as e: print(...) 在 _on_series_selection_changed 中。
问题:这会吞掉所有错误,包括代码写错的 AttributeError 或 NameError。这会让调试变得非常困难,你可能看着界面没反应,却不知道是因为数据库错了还是变量名写错了。
建议:只捕获预期的异常(如 PeeweeException),或者在捕获 Exception 后使用 traceback.print_exc() 打印完整堆栈。
4. UI 逻辑问题 (bldc_page.py)
A. 硬编码的字段映射
问题:bldc_page.py 中有大量的硬编码列表:
self.core_params_fields = [("额定转速(Rpm)", "rated_speed"), ...]同时 model.py 里也有字段定义。
风险:如果你在数据库里改了字段名(比如 rated_speed 改为 speed_rated),你需要修改 model.py 和 bldc_page.py 等多个地方。
建议:虽然完全动态化很难,但建议将这些字段映射配置(Config)提取到一个单独的常量文件(如 constants.py),让 Model 和 View 引用同一个常量定义,减少“魔法字符串”。
B. display_design_target 的字典合并逻辑
detail_object = product.detail[0] if product.detail else None
product_dict = dict(product.__data__)
combined_dict = {**product_dict, **detail_dict}
问题:product.__data__ 是 Peewee 的内部属性,直接访问它是不推荐的,因为它可能不包含继承字段或外键对象的引用。
建议:使用 model_to_dict(product, recurse=False) 更规范,或者直接访问属性 getattr(product, key)。因为你后面是用 key 来遍历 map 的,直接 getattr 其实更安全且支持 IDE 提示。
5. 总结与改进优先级
高优先级(Bug 修复):
中优先级(架构优化):
低优先级(代码整洁):
你的代码整体逻辑已经跑通了,特别是解决了之前 ProductDetail 不存在导致报错的问题(在 Model 层加了自动创建)。按照上述建议微调后,系统会更加健壮。
在 Python 桌面开发领域(PyQt/PySide),不像 Web 开发(Django/FastAPI)那样有一个统治级的“全家桶”框架。
但是,针对你目前 MVC 耦合度高、数据流混乱、逻辑堆积在 Controller 的问题,最成熟、最符合工业界标准的解决方案不是引入一个新库,而是采用 MVVM (Model-View-ViewModel) 架构模式,配合 Repository 模式。
如果非要推荐一个第三方库来辅助,我强烈推荐使用 qt_material 或 PyQt-Fluent-Widgets 来美化 UI,但核心架构必须重构。
以下是为你量身定制的 MVVM 重构蓝图。
🚀 推荐架构:MVVM + Repository Pattern
这个架构的目标是:View (界面) 只负责显示,Model (数据库) 只负责存取,ViewModel (中间层) 负责业务逻辑和状态同步。
1. 新的项目结构
my_app/
│
├── data/
│ ├── db_manager.py # 数据库连接单例
│ ├── models.py # Peewee 模型定义 (Product, ProductDetail)
│ └── repositories.py # ✅ 新增:负责所有数据库 CRUD 操作,隔离 SQL 逻辑
│
├── viewmodels/ # ✅ 新增:业务逻辑层
│ ├── base_viewmodel.py
│ ├── customer_vm.py # 处理客户/产品选择逻辑
│ └── simulation_vm.py # 处理 Maxwell 仿真逻辑 (原 DataService)
│
├── views/ # 界面层 (只包含 UI 代码)
│ ├── main_window.py
│ ├── components/ # 自定义组件
│ └── pages/
│ ├── customer_page.py
│ └── bldc_page.py
│
├── services/ # 后台服务
│ └── maxwell_worker.py # 线程任务
│
└── main.py # 程序入口,负责依赖注入
🛠️ 第一步:Repository 模式 (解决 Model 层混乱)
痛点解决:你现在的 model.py 既有类定义又有函数逻辑(find_product_by_id),容易膨胀。
重构:创建一个 ProductRepository,专门管理数据读写。
from typing import Optional
from .models import Product, ProductDetail, ProductSeries, db
class ProductRepository:
"""仓库层:只负责和数据库对话,不包含任何 UI 逻辑"""
def get_by_id(self, product_id: int) -> Optional[Product]:
try:
return (Product.select(Product, ProductDetail)
.join(ProductDetail, join_type=ProductDetail.LEFT_OUTER)
.where(Product.id == product_id)
.get())
except Product.DoesNotExist:
return None
def create(self, series: ProductSeries, model_number: str) -> Product:
with db.atomic():
product = Product.create(series=series, model_number=model_number)
ProductDetail.create(product=product)
return product
def update(self, product_id: int, data: dict):
"""统一的更新接口"""
pass
def delete(self, product_id: int):
Product.delete().where(Product.id == product_id).execute()
🧠 第二步:ViewModel 层 (解决 Controller 臃肿)
痛点解决:你的 controller.py 管得太宽了,既要知道 ComboBox 怎么选,又要处理数据库。
重构:ViewModel 持有数据状态,View 通过信号监听 ViewModel 的变化。
from PyQt6.QtCore import QObject, pyqtSignal
from data.repositories import ProductRepository
class CustomerViewModel(QObject):
series_list_loaded = pyqtSignal(list)
product_list_loaded = pyqtSignal(list)
current_product_changed = pyqtSignal(object)
status_message = pyqtSignal(str)
error_occurred = pyqtSignal(str)
def __init__(self, repo: ProductRepository):
super().__init__()
self._repo = repo
self._current_product = None
def load_initial_data(self):
"""加载初始数据"""
series = self._repo.get_all_series()
self.series_list_loaded.emit(series)
def select_series(self, series_id):
"""用户选了系列,VM 负责去查产品列表"""
if series_id == 0:
products = self._repo.get_all()
else:
products = self._repo.get_by_series(series_id)
self.product_list_loaded.emit(products)
def select_product(self, product_id):
"""用户选了产品,VM 负责查详情"""
if not product_id:
self._current_product = None
self.current_product_changed.emit(None)
return
product = self._repo.get_by_id(product_id)
self._current_product = product
self.current_product_changed.emit(product)
self.status_message.emit(f"已加载: {product.model_number}")
def save_changes(self, form_data: dict):
"""保存逻辑"""
if not self._current_product:
self.error_occurred.emit("未选择产品")
return
try:
self._repo.update(self._current_product.id, form_data)
self.status_message.emit("保存成功")
except Exception as e:
self.error_occurred.emit(str(e))
👁️ 第三步:View 层 (彻底解耦)
痛点解决:bldc_page.py 不再直接操作数据库,也不包含 DataService 逻辑。它变得非常“傻”,只负责展示和发信号。
from PyQt6.QtWidgets import QWidget, QComboBox, ...
from viewmodels.customer_vm import CustomerViewModel
class CustomerPage(QWidget):
def __init__(self, viewModel: CustomerViewModel):
super().__init__()
self._vm = viewModel
self._setup_ui()
self._bind_view_model()
def _setup_ui(self):
self.series_combo = QComboBox()
self.product_combo = QComboBox()
self.series_combo.currentIndexChanged.connect(self._on_series_changed)
def _bind_view_model(self):
"""MVVM 的核心:数据绑定"""
self._vm.series_list_loaded.connect(self._update_series_list)
self._vm.product_list_loaded.connect(self._update_product_list)
self._vm.current_product_changed.connect(self._display_product_data)
self._vm.error_occurred.connect(lambda msg: QMessageBox.critical(self, "错误", msg))
def _on_series_changed(self, index):
series_id = self.series_combo.currentData()
self._vm.select_series(series_id)
def _update_series_list(self, series_list):
self.series_combo.clear()
for s in series_list:
self.series_combo.addItem(s.name, s.id)
def _display_product_data(self, product):
if product is None:
self._clear_form()
return
self.model_number_edit.setText(product.model_number)
💡 如何解决你的“DataService 数据不一致”问题?
在 MVVM 架构中,SimulationViewModel 将替代你目前的 DataService。
SimulationViewModel 内部持有一个字典(即原来的 DataService)。
当用户在界面修改输入框时,界面发送信号 paramChanged(key, value)。
SimulationViewModel 接收信号,更新内部字典。
当切换产品时,CustomerViewModel 会通知 SimulationViewModel:“嘿,换人了”。
SimulationViewModel 立即重置内部字典,并从数据库加载新数据覆盖。
这就形成了一个单一数据源 (Single Source of Truth)。
🚀 总结:重构路线图
创建 repositories.py:把 model.py 里的 find_..., create_..., update_... 全部移进去,并加上异常处理和 try-except 保护(如自动创建 Detail)。
创建 viewmodels:
改造 bldc_page.py:
改造 main.py:
这种架构虽然前期代码量稍微多一点(多了文件),但是逻辑极其清晰。当你的 App 变得越来越复杂,有几十个页面时,你依然能轻松维护。
这是一个非常棒的决定!使用 MVVM + Repository 模式重构将彻底解决你遇到的“数据不同步”、“空指针报错”以及“代码耦合”问题。
由于我无法直接接收你的文件包,我将基于你之前提供的代码片段,为你构建一个完整的、经过重构的核心代码骨架。
你可以直接在你的项目中建立文件夹,复制粘贴这些代码。
📂 推荐的新项目结构
请按照这个结构整理你的文件:
src/
├── data/
│ ├── __init__.py
│ ├── database.py # 数据库连接与模型定义 (原 model.py)
│ └── repository.py # ✅ 核心:负责所有增删改查,隔离 SQL 细节
├── viewmodels/
│ ├── __init__.py
│ └── product_vm.py # ✅ 核心:业务逻辑,View 和 Repository 的桥梁
├── views/
│ ├── __init__.py
│ └── bldc_page.py # ✅ 界面:只负责显示,不处理逻辑
└── main.py # 程序入口,依赖注入
1. 数据层 (Data Layer)
src/data/database.py
保持模型定义纯净,去掉所有业务逻辑函数。
import os
from datetime import datetime
from peewee import (
Model, SqliteDatabase, CharField, TextField, IntegerField,
FloatField, DateTimeField, ForeignKeyField
)
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
db_path = os.path.join(base_dir, "..", "products_final.db")
db = SqliteDatabase(db_path, pragmas={'foreign_keys': 1})
class BaseModel(Model):
class Meta:
database = db
class ProductSeries(BaseModel):
name = CharField(unique=True)
description = TextField(null=True, default="")
class Product(BaseModel):
series = ForeignKeyField(ProductSeries, backref='products', null=True, on_delete='SET NULL')
model_number = CharField(unique=True)
rated_voltage = FloatField(null=True)
rated_speed = IntegerField(null=True)
rated_torque = FloatField(null=True)
created_at = DateTimeField(default=datetime.now)
class ProductDetail(BaseModel):
product = ForeignKeyField(Product, backref='detail', unique=True, on_delete='CASCADE')
price = FloatField(null=True)
poles = IntegerField(null=True)
notes = TextField(null=True, default="")
def initialize_db():
db.connect(reuse_if_open=True)
db.create_tables([ProductSeries, Product, ProductDetail], safe=True)
if ProductSeries.select().count() == 0:
ProductSeries.create(name="默认系列", description="系统自动创建")
def close_db():
if not db.is_closed():
db.close()
src/data/repository.py (关键文件)
这是解决你之前报错的“门神”。它确保了无论数据库里有没有 Detail,代码都不会崩。
import logging
from typing import Optional, Dict, Any
from peewee import IntegrityError, JOIN
from .database import db, Product, ProductDetail, ProductSeries
class ProductRepository:
"""
仓库模式:负责所有直接的数据库操作。
UI 和 ViewModel 不应该知道 SQL 或 Peewee 的存在。
"""
def get_all_series(self):
return list(ProductSeries.select())
def get_products_by_series(self, series_id: Optional[int]):
query = Product.select()
if series_id:
query = query.where(Product.series == series_id)
return list(query.order_by(Product.model_number))
def get_product_by_id(self, product_id: int) -> Optional[Product]:
"""
获取产品及其详情。
✅ 修复点:安全处理 Detail 缺失的情况。
"""
try:
product = (Product.select(Product, ProductDetail)
.join(ProductDetail, JOIN.LEFT_OUTER)
.where(Product.id == product_id)
.get())
try:
if product.detail is None:
raise ProductDetail.DoesNotExist
except ProductDetail.DoesNotExist:
logging.warning(f"Product {product_id} missing detail. Creating default.")
ProductDetail.create(product=product)
return self.get_product_by_id(product_id)
return product
except Product.DoesNotExist:
return None
def update_product(self, product_id: int, updates: Dict[str, Any]):
"""
更新产品。
✅ 修复点:使用 _meta.fields 精确匹配字段。
"""
product = self.get_product_by_id(product_id)
if not product:
raise ValueError("Product not found")
detail = product.detail
product_dirty = False
detail_dirty = False
p_fields = Product._meta.fields
d_fields = ProductDetail._meta.fields
for key, value in updates.items():
if key in ('id', 'created_at', 'series', 'product'): continue
if key in p_fields:
if str(getattr(product, key)) != str(value):
setattr(product, key, value)
product_dirty = True
elif key in d_fields:
if str(getattr(detail, key)) != str(value):
setattr(detail, key, value)
detail_dirty = True
with db.atomic():
if product_dirty: product.save()
if detail_dirty: detail.save()
return product
2. 逻辑层 (ViewModel Layer)
src/viewmodels/product_vm.py
这是 MVVM 的核心。它持有 UI 的状态,并处理业务逻辑。
from PyQt6.QtCore import QObject, pyqtSignal
from src.data.repository import ProductRepository
class ProductViewModel(QObject):
series_loaded = pyqtSignal(list)
products_loaded = pyqtSignal(list)
current_product_changed = pyqtSignal(object)
status_message = pyqtSignal(str)
error_occurred = pyqtSignal(str)
def __init__(self):
super().__init__()
self._repo = ProductRepository()
self._current_product_id = None
def load_initial_data(self):
"""初始化加载系列"""
series = self._repo.get_all_series()
self.series_loaded.emit(series)
self.load_products_by_series(None)
def load_products_by_series(self, series_id):
"""根据系列加载产品列表"""
products = self._repo.get_products_by_series(series_id)
self.products_loaded.emit(products)
def select_product(self, product_id):
"""
用户选中了某个产品 ID
"""
self._current_product_id = product_id
if product_id is None:
self.current_product_changed.emit(None)
return
product = self._repo.get_product_by_id(product_id)
if product:
self.current_product_changed.emit(product)
self.status_message.emit(f"已加载: {product.model_number}")
else:
self.error_occurred.emit("无法加载产品数据")
def save_current_product(self, form_data: dict):
"""
保存当前产品的修改
"""
if not self._current_product_id:
self.error_occurred.emit("未选择任何产品")
return
try:
self._repo.update_product(self._current_product_id, form_data)
self.status_message.emit("保存成功!")
self.select_product(self._current_product_id)
except Exception as e:
self.error_occurred.emit(f"保存失败: {str(e)}")
3. 界面层 (View Layer)
src/views/bldc_page.py
彻底解耦。它不知道数据库,也不知道 Repository,只认识 ViewModel。
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QGridLayout,
QLabel, QLineEdit, QPushButton, QComboBox, QMessageBox
)
from PyQt6.QtCore import pyqtSlot
from src.viewmodels.product_vm import ProductViewModel
class BLDCPage(QWidget):
def __init__(self, viewModel: ProductViewModel):
super().__init__()
self._vm = viewModel
self.fields_map = {}
self._setup_ui()
self._bind_view_model()
def _setup_ui(self):
layout = QVBoxLayout(self)
top_layout = QHBoxLayout()
self.series_combo = QComboBox()
self.product_combo = QComboBox()
self.save_btn = QPushButton("保存修改")
top_layout.addWidget(QLabel("系列:"))
top_layout.addWidget(self.series_combo)
top_layout.addWidget(QLabel("产品:"))
top_layout.addWidget(self.product_combo)
top_layout.addWidget(self.save_btn)
layout.addLayout(top_layout)
form_group = QGroupBox("核心参数")
grid = QGridLayout(form_group)
fields = [
("型号", "model_number"),
("额定电压(V)", "rated_voltage"),
("额定转速(rpm)", "rated_speed"),
("价格($)", "price"),
("备注", "notes")
]
for i, (label, key) in enumerate(fields):
grid.addWidget(QLabel(label), i, 0)
edit = QLineEdit()
self.fields_map[key] = edit
grid.addWidget(edit, i, 1)
layout.addWidget(form_group)
layout.addStretch()
self.series_combo.currentIndexChanged.connect(self._on_series_changed)
self.product_combo.currentIndexChanged.connect(self._on_product_changed)
self.save_btn.clicked.connect(self._on_save_clicked)
def _bind_view_model(self):
"""绑定 VM 信号到 UI 更新槽函数"""
self._vm.series_loaded.connect(self._update_series_list)
self._vm.products_loaded.connect(self._update_product_list)
self._vm.current_product_changed.connect(self._display_product)
self._vm.error_occurred.connect(lambda msg: QMessageBox.critical(self, "错误", msg))
self._vm.load_initial_data()
def _on_series_changed(self):
sid = self.series_combo.currentData()
self._vm.load_products_by_series(sid)
def _on_product_changed(self):
pid = self.product_combo.currentData()
self._vm.select_product(pid)
def _on_save_clicked(self):
data = {}
for key, widget in self.fields_map.items():
val = widget.text().strip()
if val: data[key] = val
self._vm.save_current_product(data)
@pyqtSlot(list)
def _update_series_list(self, series_list):
self.series_combo.blockSignals(True)
self.series_combo.clear()
self.series_combo.addItem("所有系列", None)
for s in series_list:
self.series_combo.addItem(s.name, s.id)
self.series_combo.blockSignals(False)
@pyqtSlot(list)
def _update_product_list(self, product_list):
self.product_combo.blockSignals(True)
self.product_combo.clear()
self.product_combo.addItem("请选择...", None)
for p in product_list:
self.product_combo.addItem(p.model_number, p.id)
self.product_combo.blockSignals(False)
@pyqtSlot(object)
def _display_product(self, product):
"""将 Product 对象的数据填入表单"""
if not product:
for widget in self.fields_map.values(): widget.clear()
return
detail = product.detail
for key, widget in self.fields_map.items():
val = None
if hasattr(product, key):
val = getattr(product, key)
elif hasattr(detail, key):
val = getattr(detail, key)
widget.setText(str(val) if val is not None else "")
4. 程序入口
src/main.py
import sys
from PyQt6.QtWidgets import QApplication, QMainWindow, QLabel
from src.data.database import initialize_db, close_db
from src.viewmodels.product_vm import ProductViewModel
from src.views.bldc_page import BLDCPage
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("BLDC Motor Manager (MVVM Refactored)")
self.resize(800, 600)
self.product_vm = ProductViewModel()
self.main_view = BLDCPage(self.product_vm)
self.setCentralWidget(self.main_view)
self.status_label = QLabel("就绪")
self.statusBar().addWidget(self.status_label)
self.product_vm.status_message.connect(self.status_label.setText)
if __name__ == '__main__':
initialize_db()
app = QApplication(sys.argv)
window = MainWindow()
window.show()
try:
sys.exit(app.exec())
finally:
close_db()
✅ 为什么这个新版本更好?
Repository 处理了“脏活累活”:repository.py 里的 get_product_by_id 自动检测并修复缺失的 ProductDetail。你再也不会看到 product.detail.get() 报错了。
ViewModel 统一了状态:UI 不再自己瞎改数据,而是收集数据传给 VM,VM 保存成功后再通知 UI 刷新。这保证了界面和数据库永远同步。
UI 极其清爽:BLDCPage 里没有一行 SQL 代码,也没有复杂的逻辑判断,只负责“显示”和“发信号”。
易于扩展:如果你想加一个新的仿真功能,只需写一个 SimulationViewModel,并在 View 里绑定它即可,完全不会影响现有的数据库逻辑。
这是一个非常棒的决定!使用 MVVM + Repository 模式重构将彻底解决你遇到的“数据不同步”、“空指针报错”以及“代码耦合”问题。
由于我无法直接接收你的文件包,我将基于你之前提供的代码片段,为你构建一个完整的、经过重构的核心代码骨架。
你可以直接在你的项目中建立文件夹,复制粘贴这些代码。
📂 推荐的新项目结构
请按照这个结构整理你的文件:
src/
├── data/
│ ├── __init__.py
│ ├── database.py # 数据库连接与模型定义 (原 model.py)
│ └── repository.py # ✅ 核心:负责所有增删改查,隔离 SQL 细节
├── viewmodels/
│ ├── __init__.py
│ └── product_vm.py # ✅ 核心:业务逻辑,View 和 Repository 的桥梁
├── views/
│ ├── __init__.py
│ └── bldc_page.py # ✅ 界面:只负责显示,不处理逻辑
└── main.py # 程序入口,依赖注入
1. 数据层 (Data Layer)
src/data/database.py
保持模型定义纯净,去掉所有业务逻辑函数。
import os
from datetime import datetime
from peewee import (
Model, SqliteDatabase, CharField, TextField, IntegerField,
FloatField, DateTimeField, ForeignKeyField
)
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
db_path = os.path.join(base_dir, "..", "products_final.db")
db = SqliteDatabase(db_path, pragmas={'foreign_keys': 1})
class BaseModel(Model):
class Meta:
database = db
class ProductSeries(BaseModel):
name = CharField(unique=True)
description = TextField(null=True, default="")
class Product(BaseModel):
series = ForeignKeyField(ProductSeries, backref='products', null=True, on_delete='SET NULL')
model_number = CharField(unique=True)
rated_voltage = FloatField(null=True)
rated_speed = IntegerField(null=True)
rated_torque = FloatField(null=True)
created_at = DateTimeField(default=datetime.now)
class ProductDetail(BaseModel):
product = ForeignKeyField(Product, backref='detail', unique=True, on_delete='CASCADE')
price = FloatField(null=True)
poles = IntegerField(null=True)
notes = TextField(null=True, default="")
def initialize_db():
db.connect(reuse_if_open=True)
db.create_tables([ProductSeries, Product, ProductDetail], safe=True)
if ProductSeries.select().count() == 0:
ProductSeries.create(name="默认系列", description="系统自动创建")
def close_db():
if not db.is_closed():
db.close()
src/data/repository.py (关键文件)
这是解决你之前报错的“门神”。它确保了无论数据库里有没有 Detail,代码都不会崩。
import logging
from typing import Optional, Dict, Any
from peewee import IntegrityError, JOIN
from .database import db, Product, ProductDetail, ProductSeries
class ProductRepository:
"""
仓库模式:负责所有直接的数据库操作。
UI 和 ViewModel 不应该知道 SQL 或 Peewee 的存在。
"""
def get_all_series(self):
return list(ProductSeries.select())
def get_products_by_series(self, series_id: Optional[int]):
query = Product.select()
if series_id:
query = query.where(Product.series == series_id)
return list(query.order_by(Product.model_number))
def get_product_by_id(self, product_id: int) -> Optional[Product]:
"""
获取产品及其详情。
✅ 修复点:安全处理 Detail 缺失的情况。
"""
try:
product = (Product.select(Product, ProductDetail)
.join(ProductDetail, JOIN.LEFT_OUTER)
.where(Product.id == product_id)
.get())
try:
if product.detail is None:
raise ProductDetail.DoesNotExist
except ProductDetail.DoesNotExist:
logging.warning(f"Product {product_id} missing detail. Creating default.")
ProductDetail.create(product=product)
return self.get_product_by_id(product_id)
return product
except Product.DoesNotExist:
return None
def update_product(self, product_id: int, updates: Dict[str, Any]):
"""
更新产品。
✅ 修复点:使用 _meta.fields 精确匹配字段。
"""
product = self.get_product_by_id(product_id)
if not product:
raise ValueError("Product not found")
detail = product.detail
product_dirty = False
detail_dirty = False
p_fields = Product._meta.fields
d_fields = ProductDetail._meta.fields
for key, value in updates.items():
if key in ('id', 'created_at', 'series', 'product'): continue
if key in p_fields:
if str(getattr(product, key)) != str(value):
setattr(product, key, value)
product_dirty = True
elif key in d_fields:
if str(getattr(detail, key)) != str(value):
setattr(detail, key, value)
detail_dirty = True
with db.atomic():
if product_dirty: product.save()
if detail_dirty: detail.save()
return product
2. 逻辑层 (ViewModel Layer)
src/viewmodels/product_vm.py
这是 MVVM 的核心。它持有 UI 的状态,并处理业务逻辑。
from PyQt6.QtCore import QObject, pyqtSignal
from src.data.repository import ProductRepository
class ProductViewModel(QObject):
series_loaded = pyqtSignal(list)
products_loaded = pyqtSignal(list)
current_product_changed = pyqtSignal(object)
status_message = pyqtSignal(str)
error_occurred = pyqtSignal(str)
def __init__(self):
super().__init__()
self._repo = ProductRepository()
self._current_product_id = None
def load_initial_data(self):
"""初始化加载系列"""
series = self._repo.get_all_series()
self.series_loaded.emit(series)
self.load_products_by_series(None)
def load_products_by_series(self, series_id):
"""根据系列加载产品列表"""
products = self._repo.get_products_by_series(series_id)
self.products_loaded.emit(products)
def select_product(self, product_id):
"""
用户选中了某个产品 ID
"""
self._current_product_id = product_id
if product_id is None:
self.current_product_changed.emit(None)
return
product = self._repo.get_product_by_id(product_id)
if product:
self.current_product_changed.emit(product)
self.status_message.emit(f"已加载: {product.model_number}")
else:
self.error_occurred.emit("无法加载产品数据")
def save_current_product(self, form_data: dict):
"""
保存当前产品的修改
"""
if not self._current_product_id:
self.error_occurred.emit("未选择任何产品")
return
try:
self._repo.update_product(self._current_product_id, form_data)
self.status_message.emit("保存成功!")
self.select_product(self._current_product_id)
except Exception as e:
self.error_occurred.emit(f"保存失败: {str(e)}")
3. 界面层 (View Layer)
src/views/bldc_page.py
彻底解耦。它不知道数据库,也不知道 Repository,只认识 ViewModel。
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QGroupBox, QGridLayout,
QLabel, QLineEdit, QPushButton, QComboBox, QMessageBox
)
from PyQt6.QtCore import pyqtSlot
from src.viewmodels.product_vm import ProductViewModel
class BLDCPage(QWidget):
def __init__(self, viewModel: ProductViewModel):
super().__init__()
self._vm = viewModel
self.fields_map = {}
self._setup_ui()
self._bind_view_model()
def _setup_ui(self):
layout = QVBoxLayout(self)
top_layout = QHBoxLayout()
self.series_combo = QComboBox()
self.product_combo = QComboBox()
self.save_btn = QPushButton("保存修改")
top_layout.addWidget(QLabel("系列:"))
top_layout.addWidget(self.series_combo)
top_layout.addWidget(QLabel("产品:"))
top_layout.addWidget(self.product_combo)
top_layout.addWidget(self.save_btn)
layout.addLayout(top_layout)
form_group = QGroupBox("核心参数")
grid = QGridLayout(form_group)
fields = [
("型号", "model_number"),
("额定电压(V)", "rated_voltage"),
("额定转速(rpm)", "rated_speed"),
("价格($)", "price"),
("备注", "notes")
]
for i, (label, key) in enumerate(fields):
grid.addWidget(QLabel(label), i, 0)
edit = QLineEdit()
self.fields_map[key] = edit
grid.addWidget(edit, i, 1)
layout.addWidget(form_group)
layout.addStretch()
self.series_combo.currentIndexChanged.connect(self._on_series_changed)
self.product_combo.currentIndexChanged.connect(self._on_product_changed)
self.save_btn.clicked.connect(self._on_save_clicked)
def _bind_view_model(self):
"""绑定 VM 信号到 UI 更新槽函数"""
self._vm.series_loaded.connect(self._update_series_list)
self._vm.products_loaded.connect(self._update_product_list)
self._vm.current_product_changed.connect(self._display_product)
self._vm.error_occurred.connect(lambda msg: QMessageBox.critical(self, "错误", msg))
self._vm.load_initial_data()
def _on_series_changed(self):
sid = self.series_combo.currentData()
self._vm.load_products_by_series(sid)
def _on_product_changed(self):
pid = self.product_combo.currentData()
self._vm.select_product(pid)
def _on_save_clicked(self):
data = {}
for key, widget in self.fields_map.items():
val = widget.text().strip()
if val: data[key] = val
self._vm.save_current_product(data)
@pyqtSlot(list)
def _update_series_list(self, series_list):
self.series_combo.blockSignals(True)
self.series_combo.clear()
self.series_combo.addItem("所有系列", None)
for s in series_list:
self.series_combo.addItem(s.name, s.id)
self.series_combo.blockSignals(False)
@pyqtSlot(list)
def _update_product_list(self, product_list):
self.product_combo.blockSignals(True)
self.product_combo.clear()
self.product_combo.addItem("请选择...", None)
for p in product_list:
self.product_combo.addItem(p.model_number, p.id)
self.product_combo.blockSignals(False)
@pyqtSlot(object)
def _display_product(self, product):
"""将 Product 对象的数据填入表单"""
if not product:
for widget in self.fields_map.values(): widget.clear()
return
detail = product.detail
for key, widget in self.fields_map.items():
val = None
if hasattr(product, key):
val = getattr(product, key)
elif hasattr(detail, key):
val = getattr(detail, key)
widget.setText(str(val) if val is not None else "")
4. 程序入口
src/main.py
我已经仔细阅读了你提供的所有源代码。代码量很大,逻辑比较完整,但正如之前分析的,存在**Controller 负担过重**、**DataService 与 Database 数据双重来源**、**View 直接处理逻辑**等架构问题。
既然你已经提供了源码,我不需要你再提供任何东西了。我将基于你现有的代码,为你生成一套**核心的重构代码**。
**注意:** 为了保持回答的长度可读且可执行,我不会复制粘贴原本用于“创建布局”的几百行 UI 代码(例如 `_create_form_grid`),因为那些不需要改。我主要重写**数据流向**和**逻辑控制**的部分。
请按照以下步骤,在你的项目中创建文件夹并填入代码。
---
请务必按照这个结构来组织文件:
```text
src/
├── main.py
├── data/
│ ├── __init__.py
│ ├── model.py
│ └── repository.py
├── viewmodels/
│ ├── __init__.py
│ ├── product_vm.py
│ └── simulation_vm.py
├── views/
│ ├── __init__.py
│ ├── main_view.py
│ ├── design_target_page.py
│ ├── bldc_page.py
│ └── product_detail_page.py
└── utils/
├── worker.py
├── normalize.py
└── maxwell_modules/ ...
```
---
**操作**:保留你原来的 `BaseModel`, `ProductSeries`, `Product`, `ProductDetail`, `initialize_database`, `close_connection`。
**删除**:删除文件下方所有的 `find_product_by_id`, `create_full_product` 等函数,因为我们要把它们移到 `repository.py` 中。
这是解决空指针报错和统一数据访问的关键。
```python
import logging
from peewee import JOIN, IntegrityError, DoesNotExist
from .model import db, Product, ProductDetail, ProductSeries
class ProductRepository:
"""负责所有数据库交互,View 和 ViewModel 不直接操作 Model"""
def get_all_series(self):
return list(ProductSeries.select().order_by(ProductSeries.name))
def get_products_by_series(self, series_id):
query = Product.select()
if series_id and series_id != 0:
query = query.where(Product.series == series_id)
return list(query.order_by(Product.model_number))
def get_product_by_id(self, product_id: int):
"""获取产品,如果详情不存在则自动修复"""
try:
product = (Product.select(Product, ProductDetail)
.join(ProductDetail, JOIN.LEFT_OUTER)
.where(Product.id == product_id)
.get())
try:
if product.detail is None:
raise DoesNotExist
except (DoesNotExist, AttributeError):
logging.warning(f"Product {product_id} missing detail. Auto-creating.")
ProductDetail.create(product=product)
return self.get_product_by_id(product_id)
return product
except DoesNotExist:
return None
def create_product(self, series_id: int, model_number: str):
"""创建新产品"""
series = ProductSeries.get_by_id(series_id)
with db.atomic():
product = Product.create(series=series, model_number=model_number)
ProductDetail.create(product=product)
return product
def update_product(self, product_id: int, updates: dict):
"""更新产品,使用 _meta.fields 过滤字段"""
product = self.get_product_by_id(product_id)
if not product: return False
detail = product.detail
p_dirty, d_dirty = False, False
p_fields = Product._meta.fields
d_fields = ProductDetail._meta.fields
for key, value in updates.items():
if key in ('id', 'created_at', 'series', 'product'): continue
str_val = str(value) if value is not None else ""
if key in p_fields:
old_val = str(getattr(product, key)) if getattr(product, key) is not None else ""
if old_val != str_val:
setattr(product, key, value)
p_dirty = True
elif key in d_fields:
old_val = str(getattr(detail, key)) if getattr(detail, key) is not None else ""
if old_val != str_val:
setattr(detail, key, value)
d_dirty = True
with db.atomic():
if p_dirty: product.save()
if d_dirty: detail.save()
return True
def delete_product(self, product_id: int):
Product.delete().where(Product.id == product_id).execute()
```
---
这个类替代了 `Controller` 中关于“客户需求页”的所有逻辑。
```python
from PyQt6.QtCore import QObject, pyqtSignal
from src.data.repository import ProductRepository
from src.data.model import Product
class ProductViewModel(QObject):
series_list_updated = pyqtSignal(list)
product_list_updated = pyqtSignal(list)
current_product_changed = pyqtSignal(object)
status_msg = pyqtSignal(str)
error_msg = pyqtSignal(str)
def __init__(self):
super().__init__()
self.repo = ProductRepository()
self._current_product_id = None
def load_init_data(self):
series = self.repo.get_all_series()
self.series_list_updated.emit(series)
self.filter_products_by_series(0)
def filter_products_by_series(self, series_id):
products = self.repo.get_products_by_series(series_id)
self.product_list_updated.emit(products)
def select_product(self, product_id):
self._current_product_id = product_id
if not product_id:
self.current_product_changed.emit(None)
self.status_msg.emit("就绪")
return
product = self.repo.get_product_by_id(product_id)
if product:
self.current_product_changed.emit(product)
self.status_msg.emit(f"已加载: {product.model_number}")
else:
self.error_msg.emit("加载产品失败")
def create_new_product(self, series_id, model_number):
try:
new_prod = self.repo.create_product(series_id, model_number)
self.filter_products_by_series(series_id)
self.status_msg.emit(f"产品 {model_number} 创建成功")
return new_prod.id
except Exception as e:
self.error_msg.emit(f"创建失败: {str(e)}")
return None
def save_changes(self, form_data: dict):
if not self._current_product_id:
return
try:
self.repo.update_product(self._current_product_id, form_data)
self.status_msg.emit("保存成功")
self.select_product(self._current_product_id)
except Exception as e:
self.error_msg.emit(f"保存失败: {str(e)}")
def delete_current_product(self):
if self._current_product_id:
self.repo.delete_product(self._current_product_id)
self.filter_products_by_series(0)
self.select_product(None)
self.status_msg.emit("产品已删除")
```
这个类替代了 `DataService` 和 `bldc_page` 中的逻辑部分。它实现了**单一数据源**。
```python
from PyQt6.QtCore import QObject, pyqtSignal, pyqtSlot
from src.utils.worker import MaxwellWorker
from src.utils.maxwell_modules.maxwell_config import MaxwellConfig
from src.data.model import Product
from pathlib import Path
class SimulationViewModel(QObject):
parameter_changed = pyqtSignal(str, object)
log_received = pyqtSignal(str)
progress_updated = pyqtSignal(int)
simulation_finished = pyqtSignal(object)
simulation_error = pyqtSignal(str)
def __init__(self):
super().__init__()
self.parameters = {}
self.worker = MaxwellWorker()
self.worker.log_message.connect(self.log_received)
self.worker.progress_updated.connect(self.progress_updated)
self.worker.finished.connect(self.simulation_finished)
self.worker.error.connect(self.simulation_error)
def load_product(self, product: Product):
"""当用户在主界面切换产品时调用"""
self.parameters.clear()
if not product: return
d_dict = product.detail.__data__ if product.detail else {}
p_dict = product.__data__
self.parameters.update(p_dict)
self.parameters.update(d_dict)
def update_parameter(self, key, value):
"""UI 输入框变化时调用"""
self.parameters[key] = value
def start_simulation(self, project_path, project_name):
if self.worker.isRunning():
return
config = MaxwellConfig(project_path, Path(project_path).parent,
project_name, "RMxprtDesign", "Setup")
self.worker.start()
self.worker.execute_simulation_task(config, self.parameters)
def cleanup(self):
self.worker.cleanup()
```
---
1. **构造函数**:接收 `ProductViewModel`。
2. **删除**:删除 `DataService`。
3. **连接**:
* 下拉框 `currentIndexChanged` -> 调用 `vm.select_product(id)`。
* 保存按钮 `clicked` -> 收集 `self.fields_map` 数据 -> 调用 `vm.save_changes(data)`。
* VM 信号 `current_product_changed` -> 触发 `load_data_to_form(product)`。
```python
class CustomerRequirementsPage(QWidget):
def __init__(self, viewModel):
super().__init__()
self.vm = viewModel
self.vm.product_list_updated.connect(self.update_product_combo_box)
self.vm.current_product_changed.connect(self.load_data_to_form)
self.save_button.clicked.connect(self._on_save)
def _on_save(self):
data, _ = self.get_data_from_form()
full_data = {**data[0], **data[1]}
self.vm.save_changes(full_data)
```
1. **构造函数**:接收 `SimulationViewModel`。
2. **删除**:删除 `DataService`, `MaxwellWorker` (因为移到了 VM)。
3. **修改输入框**:`textChanged` 事件不再更新 `DataService`,而是调用 `self.vm.update_parameter(key, value)`。
4. **修改开始按钮**:调用 `self.vm.start_simulation(...)`。
```python
class MaxwellApp(QWidget):
def __init__(self, sim_vm):
super().__init__()
self.vm = sim_vm
self.vm.log_received.connect(self.log_output.append)
self.vm.simulation_finished.connect(self._on_sim_finished)
def _on_text_changed(self, key, text):
self.vm.update_parameter(key, value)
def _start_simulation(self):
self.vm.start_simulation(self.selected_project_path, self.project_name)
```
---
这是组装所有组件的地方。
```python
import sys
from PyQt6.QtWidgets import QApplication
from src.data.model import initialize_database, close_connection
from src.viewmodels.product_vm import ProductViewModel
from src.viewmodels.simulation_vm import SimulationViewModel
from src.views.main_view import MainView
def main():
app = QApplication(sys.argv)
initialize_database()
product_vm = ProductViewModel()
sim_vm = SimulationViewModel()
product_vm.current_product_changed.connect(sim_vm.load_product)
main_window = MainView(product_vm, sim_vm)
main_window.show()
product_vm.load_init_data()
exit_code = app.exec()
sim_vm.cleanup()
close_connection()
sys.exit(exit_code)
if __name__ == "__main__":
main()
```
你需要稍微改一下 `MainView` 的 `__init__` 和 `_create_main_content_stack`:
```python
class MainView(QMainWindow):
def __init__(self, product_vm, sim_vm):
super().__init__()
self.product_vm = product_vm
self.sim_vm = sim_vm
self.main_content_stack = self._create_main_content_stack()
def _create_main_content_stack(self):
stack = QStackedWidget()
self.customer_page = CustomerRequirementsPage(self.product_vm)
self.bldc_page = MaxwellApp(self.sim_vm)
self.detail_page = ProductDetailPage()
stack.addWidget(self.customer_page)
stack.addWidget(self.bldc_page)
stack.addWidget(self.detail_page)
return stack
```
这套方案直接使用了你提供的 Peewee Model 定义和 PyQt 界面代码,但重构了**连接方式**。
1. **Repository** 帮你自动处理了 `product.detail` 为空的情况,消灭了空指针 BUG。
2. **ProductViewModel** 统一了数据库读写逻辑,不再依赖 `controller.py`。
3. **SimulationViewModel** 统一了仿真参数,替代了 `DataService`,解决了“数据不同步”的问题。
4. **MainView** 负责依赖注入,让各个组件松耦合。
✅ 为什么这个新版本更好?
Repository 处理了“脏活累活”:repository.py 里的 get_product_by_id 自动检测并修复缺失的 ProductDetail。你再也不会看到 product.detail.get() 报错了。
ViewModel 统一了状态:UI 不再自己瞎改数据,而是收集数据传给 VM,VM 保存成功后再通知 UI 刷新。这保证了界面和数据库永远同步。
UI 极其清爽:BLDCPage 里没有一行 SQL 代码,也没有复杂的逻辑判断,只负责“显示”和“发信号”。
易于扩展:如果你想加一个新的仿真功能,只需写一个 SimulationViewModel,并在 View 里绑定它即可,完全不会影响现有的数据库逻辑。
n5321 | 2025年11月22日 00:25