Yorusora/Signal_Label_Reborn
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1、完成<数据粗同步>的全部功能

Browse Source 
2、将业务返回结果封装成了一个Result对象
This commit is contained in:
Yorusora
2025-05-10 19:45:57 +08:00
parent 2bd2a1843f
commit b9f9122a65
22 changed files with 3821 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
func/Module_detect_Rpeak.py
View File

@ -12,6 +12,7 @@ from yaml import dump, load, FullLoader
from func.utils.PublicFunc import PublicFunc
from func.utils.Constants import Constants, ConfigParams
from func.utils.Result import Result
from func.utils.detect_Rpeak import preprocess, Rpeak_Detection, get_method
from ui.MainWindow.MainWindow_detect_Rpeak import Ui_MainWindow_detect_Rpeak
@ -183,26 +184,29 @@ class MainWindow_detect_Rpeak(QMainWindow):
@overrides
def closeEvent(self, event):
PublicFunc.__disableAllButton__(self, ButtonState)
reply = QMessageBox.question(self, '确认', '确认退出吗？', QMessageBox.Yes | QMessageBox.No, QMessageBox.No)
if reply == QMessageBox.Yes:
PublicFunc.__disableAllButton__(self, ButtonState)
PublicFunc.statusbar_show_msg(self, PublicFunc.format_status_msg(Constants.SHUTTING_DOWN))
QApplication.processEvents()
PublicFunc.statusbar_show_msg(self, PublicFunc.format_status_msg(Constants.SHUTTING_DOWN))
QApplication.processEvents()
# 清空画框
self.ax0.clear()
self.ax1.clear()
# 清空画框
self.ax0.clear()
self.ax1.clear()
# 释放资源
del self.data
self.fig.clf()
plt.close(self.fig)
self.deleteLater()
collect()
self.canvas = None
event.accept()
# 释放资源
del self.data
self.fig.clf()
plt.close(self.fig)
self.deleteLater()
collect()
self.canvas = None
event.accept()
else:
event.ignore()
@staticmethod
def __reset__():
def __reset__(self):
ButtonState["Current"].update(ButtonState["Default"].copy())
ButtonState["Current"]["pushButton_view"] = True
@ -227,14 +231,11 @@ class MainWindow_detect_Rpeak(QMainWindow):
label=Constants.DETECT_RPEAK_PLOT_LABEL_INTERVAL)
self.ax0.legend(loc=Constants.PLOT_UPPER_RIGHT)
self.ax1.legend(loc=Constants.PLOT_UPPER_RIGHT)
status = True
info = Constants.DRAWING_FINISHED
self.canvas.draw()
return Result().success(info=Constants.DRAWING_FINISHED)
else:
status = False
info = Constants.DRAWING_FAILURE
self.canvas.draw()
return status, info
self.canvas.draw()
return Result().failure(info=Constants.DRAWING_FAILURE)
def __update_config__(self):
Config["Filter"]["BandPassLow"] = self.ui.doubleSpinBox_bandPassLow.value()
@ -259,32 +260,30 @@ class MainWindow_detect_Rpeak(QMainWindow):
# TODO：获取检测方法的解耦
method_list = get_method()
if len(method_list) == 0 or method_list is None:
status = False
info = Constants.DETECT_RPEAK_LOAD_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Method_Not_Exist"]
result = Result().failure(info=Constants.DETECT_RPEAK_LOAD_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Method_Not_Exist"])
else:
status = True
info = Constants.DETECT_RPEAK_LOAD_FINISHED
if not status:
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
result = Result().success(info=Constants.DETECT_RPEAK_LOAD_FINISHED)
if not result.status:
PublicFunc.text_output(self.ui, "(1/2)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/2)" + result.info, Constants.TIPS_TYPE_INFO)
self.update_ui_comboBox_method(method_list)
# 导入数据
PublicFunc.progressbar_update(self, 2, 2, Constants.INPUTTING_DATA, 10)
status, info = self.data.open_file()
if not status:
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
result = self.data.open_file()
if not result.status:
PublicFunc.text_output(self.ui, "(2/2)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(2/2)" + result.info, Constants.TIPS_TYPE_INFO)
MainWindow_detect_Rpeak.__reset__()
self.__reset__()
PublicFunc.finish_operation(self, ButtonState)
def __slot_btn_view__(self):
@ -292,25 +291,25 @@ class MainWindow_detect_Rpeak(QMainWindow):
# 数据预处理
PublicFunc.progressbar_update(self, 1, 3, Constants.DETECT_RPEAK_PROCESSING_DATA, 0)
status, info = self.data.preprocess()
if not status:
PublicFunc.text_output(self.ui, "(1/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
result = self.data.preprocess()
if not result.status:
PublicFunc.text_output(self.ui, "(1/3)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(1/3)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/3)" + result.info, Constants.TIPS_TYPE_INFO)
# 预测峰值
PublicFunc.progressbar_update(self, 2, 3, Constants.DETECT_RPEAK_PREDICTING_PEAK, 10)
status, info = self.data.predict_Rpeak()
if not status:
PublicFunc.text_output(self.ui, "(2/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
result = self.data.predict_Rpeak()
if not result.status:
PublicFunc.text_output(self.ui, "(2/3)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(2/3)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(2/3)" + result.info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_RPEAK_DATA_LENGTH_POINTS + str(len(self.data.raw_data)),
Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_RPEAK_DURATION_MIN +
@ -321,14 +320,14 @@ class MainWindow_detect_Rpeak(QMainWindow):
# 绘图
PublicFunc.progressbar_update(self, 3, 3, Constants.DRAWING_DATA, 70)
status, info = self.__plot__()
if not status:
PublicFunc.text_output(self.ui, "(3/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
result = self.__plot__()
if not result.status:
PublicFunc.text_output(self.ui, "(3/3)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(3/3)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(3/3)" + result.info, Constants.TIPS_TYPE_INFO)
ButtonState["Current"]["pushButton_save"] = True
PublicFunc.finish_operation(self, ButtonState)
@ -350,20 +349,20 @@ class MainWindow_detect_Rpeak(QMainWindow):
for start in range(0, total_rows, chunk_size):
end = min(start + chunk_size, total_rows)
chunk = DataFrame(self.data.peak.reshape(-1)).iloc[start:end]
status, info = self.data.save(chunk)
result = self.data.save(chunk)
progress = int((end / total_rows) * 100)
self.progressbar.setValue(progress)
QApplication.processEvents()
if not status:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
if not result.status:
PublicFunc.text_output(self.ui, "(1/1)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
PublicFunc.finish_operation(self, ButtonState)
return
else:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/1)" + result.info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, result.info, Constants.TIPS_TYPE_INFO)
PublicFunc.finish_operation(self, ButtonState)
def reset_axes(self):
@ -391,20 +390,20 @@ class Data:
def open_file(self):
if not Path(Config["Path"]["Input"]).exists():
return False, Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Data_Path_Not_Exist"]
return Result().failure(info=Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Data_Path_Not_Exist"])
try:
self.raw_data = read_csv(Config["Path"]["Input"],
encoding=ConfigParams.UTF8_ENCODING,
header=None).to_numpy().reshape(-1)
except Exception:
return False, Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Read_Data_Exception"]
return Result().failure(info=Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Read_Data_Exception"])
return True, Constants.INPUT_FINISHED
return Result().success(info=Constants.INPUT_FINISHED)
def preprocess(self):
if self.raw_data is None:
return False, Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Raw_Data_Not_Exist"]
return Result().failure(info=Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Raw_Data_Not_Exist"])
try:
self.processed_data = preprocess(self.raw_data,
@ -412,13 +411,13 @@ class Data:
Config["Filter"]["BandPassLow"],
Config["Filter"]["BandPassHigh"])
except Exception:
return False, Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Filter_Exception"]
return Result().failure(info=Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Filter_Exception"])
return True, Constants.DETECT_RPEAK_PROCESS_FINISHED
return Result().success(info=Constants.DETECT_RPEAK_PROCESS_FINISHED)
def predict_Rpeak(self):
if self.processed_data is None:
return False, Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Processed_Data_Not_Exist"]
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Processed_Data_Not_Exist"])
try:
self.peak, self.interval, self.RRIV = Rpeak_Detection(self.processed_data,
@ -426,17 +425,18 @@ class Data:
Config["PeaksValue"],
Config["DetectMethod"])
except Exception:
return False, Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Predict_Exception"]
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Predict_Exception"])
return True, Constants.DETECT_RPEAK_PREDICT_FINISHED
return Result().success(info=Constants.DETECT_RPEAK_PREDICT_FINISHED)
def save(self, chunk):
if self.peak is None:
return False, Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Peak_Not_Exist"]
return Result().failure(info=Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Peak_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save"], mode='a', index=False, header=False)
except Exception:
return False, Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Save_Exception"]
return Result().failure(info=Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Save_Exception"])
return True, Constants.SAVING_FINISHED
return Result().success(info=Constants.SAVING_FINISHED)