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

1、完成了<ECG的R峰算法定位>的重构

Browse Source 
2、创建好了<人工纠正>和<体动标注>的界面绘制
This commit is contained in:
Yorusora
2025-04-28 16:18:59 +08:00
parent f928fa4d9c
commit 2a13ceac39
27 changed files with 5552 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
func/Module_detect_Jpeak.py
View File

@ -59,7 +59,6 @@ class SettingWindow(QMainWindow):
def __read_config__(self):
if not Path(ConfigParams.DETECT_JPEAK_CONFIG_FILE_PATH).exists():
with open(ConfigParams.DETECT_JPEAK_CONFIG_FILE_PATH, "w") as f:
dump(ConfigParams.DETECT_JPEAK_CONFIG_NEW_CONTENT, f)
@ -246,13 +245,13 @@ class MainWindow_detect_Jpeak(QMainWindow):
if sender == self.ui.pushButton_view:
self.line_data, = self.ax0.plot(self.data.processed_data,
color=Constants.PLOT_COLOR_BLUE,
label=Constants.DETECT_J_PEAK_PLOT_LABEL_BCG)
label=Constants.DETECT_JPEAK_PLOT_LABEL_BCG)
self.point_peak, = self.ax0.plot(self.data.peak, self.data.processed_data[self.data.peak],
'r.',
label=Constants.DETECT_J_PEAK_PLOT_LABEL_J_PEAKS)
label=Constants.DETECT_JPEAK_PLOT_LABEL_J_PEAKS)
self.line_interval, = self.ax0.plot(self.data.interval,
color=Constants.PLOT_COLOR_ORANGE,
label=Constants.DETECT_J_PEAK_PLOT_LABEL_INTERVAL)
label=Constants.DETECT_JPEAK_PLOT_LABEL_INTERVAL)
self.ax0.legend(loc=Constants.PLOT_UPPER_RIGHT)
status = True
info = Constants.DRAWING_FINISHED
@ -293,6 +292,41 @@ class MainWindow_detect_Jpeak(QMainWindow):
if widget.objectName() in ButtonState["Default"].keys():
widget.setEnabled(ButtonState["Default"][widget.objectName()])
def __update_config__(self):
Config["Filter"]["BandPassLow"] = self.ui.doubleSpinBox_bandPassLow.value()
Config["Filter"]["BandPassHigh"] = self.ui.doubleSpinBox_bandPassHigh.value()
Config["PeaksValue"] = self.ui.spinBox_peaksValue.value()
Config["AmpValue"] = self.ui.doubleSpinBox_ampValue.value()
Config["IntervalLow"] = self.ui.spinBox_intervalLow.value()
Config["IntervalHigh"] = self.ui.spinBox_intervalHigh.value()
Config["UseCPU"] = self.ui.checkBox_useCPU.isChecked()
Config["DetectMethod"] = self.ui.comboBox_model.currentText()
def finish_operation(self):
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.OPERATION_FINISHED))
self.progressbar.setValue(100)
QApplication.processEvents()
self.__enableAllButton__()
def add_progressbar(self):
self.progressbar = QProgressBar()
self.progressbar.setRange(0, 100)
self.progressbar.setValue(0)
self.progressbar.setStyleSheet(Constants.PROGRESSBAR_STYLE)
self.ui.statusbar.addPermanentWidget(self.progressbar)
def statusbar_show_msg(self, msg):
self.ui.statusbar.showMessage(msg)
def statusbar_clear_msg(self):
self.ui.statusbar.clearMessage()
def __slot_btn_input__(self):
self.__disableAllButton__()
@ -310,7 +344,7 @@ class MainWindow_detect_Jpeak(QMainWindow):
# 清空模型列表
self.ui.comboBox_model.clear()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.LOADING_MODEL))
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/2)" + Constants.DETECT_JPEAK_LOADING_MODEL))
self.progressbar.setValue(0)
QApplication.processEvents()
@ -318,15 +352,15 @@ class MainWindow_detect_Jpeak(QMainWindow):
self.model = Model()
status, info = self.model.seek_model()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_INFO)
self.update_ui_comboBox_model(self.model.model_list)
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.INPUTTING_DATA))
self.statusbar_show_msg(PublicFunc.format_status_msg("(2/2)" + Constants.INPUTTING_DATA))
self.progressbar.setValue(10)
QApplication.processEvents()
@ -334,12 +368,12 @@ class MainWindow_detect_Jpeak(QMainWindow):
self.data = Data()
status, info = self.data.open_file()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_INFO)
MainWindow_detect_Jpeak.__reset__()
self.finish_operation()
@ -348,21 +382,21 @@ class MainWindow_detect_Jpeak(QMainWindow):
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.DETECT_JPEAK_PROCESSING_DATA))
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/3)" + Constants.DETECT_JPEAK_PROCESSING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 数据预处理
status, info = self.data.preprocess()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(1/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/3)" + info, Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.DETECT_JPEAK_PREDICTING_PEAK))
self.statusbar_show_msg(PublicFunc.format_status_msg("(2/3)" + Constants.DETECT_JPEAK_PREDICTING_PEAK))
self.progressbar.setValue(10)
QApplication.processEvents()
@ -370,33 +404,33 @@ class MainWindow_detect_Jpeak(QMainWindow):
self.model.selected_model = Config["DetectMethod"]
status, info = self.data.predict_Jpeak(self.model)
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(2/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_J_PEAK_DATA_LENGTH_POINTS + str(len(self.data.data)),
PublicFunc.text_output(self.ui, "(2/3)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_JPEAK_DATA_LENGTH_POINTS + str(len(self.data.raw_data)),
Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_J_PEAK_DURATION_MIN +
str((len(self.data.data) / Config["InputConfig"]["Freq"] / 60)),
PublicFunc.text_output(self.ui, Constants.DETECT_JPEAK_DURATION_MIN +
str((len(self.data.raw_data) / Config["InputConfig"]["Freq"] / 60)),
Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_J_PEAK_JPEAK_AMOUNT + str(len(self.data.peak)),
PublicFunc.text_output(self.ui, Constants.DETECT_JPEAK_PEAK_AMOUNT + str(len(self.data.peak)),
Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.DRAWING_DATA))
self.statusbar_show_msg(PublicFunc.format_status_msg("(3/3)" + Constants.DRAWING_DATA))
self.progressbar.setValue(70)
QApplication.processEvents()
# 绘图
status, info = self.__plot__()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(3/3)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(3/3)" + info, Constants.TIPS_TYPE_INFO)
ButtonState["Current"]["pushButton_save"] = True
self.finish_operation()
@ -410,14 +444,14 @@ class MainWindow_detect_Jpeak(QMainWindow):
if reply == QMessageBox.Yes:
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.SAVING_DATA))
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/1)" + Constants.SAVING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 保存
# status, info = self.data.save()
total_rows = len(DataFrame(self.data.peak.reshape(-1)))
chunk_size = ConfigParams.PREPROCESS_SAVE_CHUNK_SIZE
chunk_size = ConfigParams.DETECT_JPEAK_SAVE_CHUNK_SIZE
with open(Config["Path"]["Save"], 'w') as f:
for start in range(0, total_rows, chunk_size):
end = min(start + chunk_size, total_rows)
@ -428,56 +462,21 @@ class MainWindow_detect_Jpeak(QMainWindow):
QApplication.processEvents()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, info, Constants.TIPS_TYPE_INFO)
self.finish_operation()
def __update_config__(self):
Config["Filter"]["BandPassLow"] = self.ui.doubleSpinBox_bandPassLow.value()
Config["Filter"]["BandPassHigh"] = self.ui.doubleSpinBox_bandPassHigh.value()
Config["PeaksValue"] = self.ui.spinBox_peaksValue.value()
Config["AmpValue"] = self.ui.doubleSpinBox_ampValue.value()
Config["IntervalLow"] = self.ui.spinBox_intervalLow.value()
Config["IntervalHigh"] = self.ui.spinBox_intervalHigh.value()
Config["UseCPU"] = self.ui.checkBox_useCPU.isChecked()
Config["DetectMethod"] = self.ui.comboBox_model.currentText()
def update_ui_comboBox_model(self, model_list):
self.ui.comboBox_model.clear()
self.ui.comboBox_model.addItems(model_list)
def finish_operation(self):
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.OPERATION_FINISHED))
self.progressbar.setValue(100)
QApplication.processEvents()
self.__enableAllButton__()
def add_progressbar(self):
self.progressbar = QProgressBar()
self.progressbar.setRange(0, 100)
self.progressbar.setValue(0)
self.progressbar.setStyleSheet(Constants.PROGRESSBAR_STYLE)
self.ui.statusBar.addPermanentWidget(self.progressbar)
def statusbar_show_msg(self, msg):
self.ui.statusBar.showMessage(msg)
def statusbar_clear_msg(self):
self.ui.statusBar.clearMessage()
class Data:
@ -486,7 +485,7 @@ class Data:
self.file_path_input = Config["Path"]["Input"]
self.file_path_save = Config["Path"]["Save"]
self.data = None
self.raw_data = None
self.processed_data = None
self.peak = None
self.interval = None
@ -497,21 +496,21 @@ class Data:
return False, Constants.INPUT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Data_Path_Not_Exist"]
try:
self.data = read_csv(self.file_path_input,
self.raw_data = read_csv(self.file_path_input,
encoding=ConfigParams.UTF8_ENCODING,
header=None).to_numpy().reshape(-1)
except Exception:
return False, Constants.INPUT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Read_File_Exception"]
return False, Constants.INPUT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Read_Data_Exception"]
return True, Constants.INPUT_FINISHED
def preprocess(self):
if self.data is None:
if self.raw_data is None:
return False, Constants.DETECT_JPEAK_PROCESS_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Raw_Data_Not_Exist"]
try:
self.processed_data = preprocess(self.data,
self.processed_data = preprocess(self.raw_data,
Config["InputConfig"]["Freq"],
Config["Filter"]["BandPassLow"],
Config["Filter"]["BandPassHigh"],
@ -555,7 +554,7 @@ class Data:
# float_format='%.4f')
chunk.to_csv(self.file_path_save, mode='a', index=False, header=False)
except Exception:
return False, Constants.SAVING_FAILURE + Constants.PREPROCESS_FAILURE_REASON["Save_Exception"]
return False, Constants.SAVING_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Save_Exception"]
return True, Constants.SAVING_FINISHED
@ -572,13 +571,13 @@ class Model:
def seek_model(self):
if not Path(Config["ModelFolderPath"]).exists():
return False, Constants.LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Model_Path_Not_Exist"]
return False, Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Model_Path_Not_Exist"]
try:
self.model_list = [file.name for file in Path(Config["ModelFolderPath"]).iterdir() if file.is_file()]
if len(self.model_list) == 0:
return False, Constants.DETECT_JPEAK_FAILURE_REASON["Model_File_Not_Exist"]
except Exception:
return False, Constants.LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Read_Model_Exception"]
return False, Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Read_Model_Exception"]
return True, Constants.LOAD_FINISHED
return True, Constants.DETECT_JPEAK_LOAD_FINISHED

550
func/Module_detect_Rpeak.py Normal file
View File

@ -0,0 +1,550 @@
from gc import collect
from pathlib import Path
import matplotlib.pyplot as plt
from PySide6.QtWidgets import QMessageBox, QMainWindow, QWidget, QPushButton, QProgressBar, QApplication
from matplotlib import gridspec
from matplotlib.backends.backend_qt import NavigationToolbar2QT
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg
from overrides import overrides
from pandas import read_csv, DataFrame
from yaml import dump, load, FullLoader
from func.utils.PublicFunc import PublicFunc
from func.utils.Constants import Constants, ConfigParams
from func.utils.detect_Rpeak import preprocess, Rpeak_Detection, get_method
from ui.MainWindow.MainWindow_detect_Rpeak import Ui_MainWindow_detect_Rpeak
from ui.setting.detect_Rpeak_input_setting import Ui_MainWindow_detect_Rpeak_input_setting
Config = {
}
ButtonState = {
"Default": {
"pushButton_input_setting": True,
"pushButton_input": True,
"pushButton_view": False,
"pushButton_save": False
},
"Current": {
"pushButton_input_setting": True,
"pushButton_input": True,
"pushButton_view": False,
"pushButton_save": False
}
}
class SettingWindow(QMainWindow):
def __init__(self, root_path, sampID):
super(SettingWindow, self).__init__()
self.ui = Ui_MainWindow_detect_Rpeak_input_setting()
self.ui.setupUi(self)
self.root_path = root_path
self.sampID = sampID
self.config = None
self.__read_config__()
self.ui.spinBox_input_freq.valueChanged.connect(self.__update_ui__)
self.ui.pushButton_confirm.clicked.connect(self.__write_config__)
self.ui.pushButton_cancel.clicked.connect(self.__rollback_config__)
self.ui.pushButton_cancel.clicked.connect(self.close)
def __read_config__(self):
if not Path(ConfigParams.DETECT_RPEAK_CONFIG_FILE_PATH).exists():
with open(ConfigParams.DETECT_RPEAK_CONFIG_FILE_PATH, "w") as f:
dump(ConfigParams.DETECT_RPEAK_CONFIG_NEW_CONTENT, f)
with open(ConfigParams.DETECT_RPEAK_CONFIG_FILE_PATH, "r") as f:
file_config = load(f.read(), Loader=FullLoader)
Config.update(file_config)
self.config = file_config
Config.update({
"Path": {
"Input": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.DETECT_RPEAK_INPUT_ECG_FILENAME +
str(Config["InputConfig"]["Freq"]) +
ConfigParams.ENDSWITH_TXT))),
"Save": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.DETECT_RPEAK_SAVE_FILENAME +
ConfigParams.ENDSWITH_TXT)))
}
})
# 数据回显
self.ui.spinBox_input_freq.setValue(Config["InputConfig"]["Freq"])
self.ui.plainTextEdit_file_path_input.setPlainText(Config["Path"]["Input"])
self.ui.plainTextEdit_file_path_save.setPlainText(Config["Path"]["Save"])
def __write_config__(self):
# 从界面写入配置
Config["InputConfig"]["Freq"] = self.ui.spinBox_input_freq.value()
Config["Path"]["Input"] = self.ui.plainTextEdit_file_path_input.toPlainText()
Config["Path"]["Save"] = self.ui.plainTextEdit_file_path_save.toPlainText()
# 保存配置到文件
self.config["InputConfig"]["Freq"] = self.ui.spinBox_input_freq.value()
with open(ConfigParams.DETECT_RPEAK_CONFIG_FILE_PATH, "w") as f:
dump(self.config, f)
self.close()
def __rollback_config__(self):
self.__read_config__()
def __update_ui__(self):
self.ui.plainTextEdit_file_path_input.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.DETECT_RPEAK_INPUT_ECG_FILENAME +
str(self.ui.spinBox_input_freq.value()) +
ConfigParams.ENDSWITH_TXT))))
class MainWindow_detect_Rpeak(QMainWindow):
def __init__(self):
super(MainWindow_detect_Rpeak, self).__init__()
self.ui = Ui_MainWindow_detect_Rpeak()
self.ui.setupUi(self)
self.root_path = None
self.sampID = None
self.data = None
self.setting = None
# 初始化进度条
self.progressbar = None
self.add_progressbar()
#初始化画框
self.fig = None
self.canvas = None
self.figToolbar = None
self.gs = None
self.ax0 = None
self.ax1 = None
self.line_data = None
self.point_peak = None
self.line_interval = None
self.point_RRIV = None
self.msgBox = QMessageBox()
self.msgBox.setWindowTitle(Constants.MAINWINDOW_MSGBOX_TITLE)
@overrides
def show(self, root_path, sampID):
super().show()
self.root_path = root_path
self.sampID = sampID
self.setting = SettingWindow(root_path, sampID)
# 初始化画框
self.fig = plt.figure(figsize=(12, 9), dpi=100)
self.canvas = FigureCanvasQTAgg(self.fig)
self.figToolbar = NavigationToolbar2QT(self.canvas)
for action in self.figToolbar.actions():
if action.text() == "Subplots" or action.text() == "Customize":
self.figToolbar.removeAction(action)
self.ui.verticalLayout_canvas.addWidget(self.canvas)
self.ui.verticalLayout_canvas.addWidget(self.figToolbar)
self.gs = gridspec.GridSpec(2, 1, height_ratios=[1, 1])
self.fig.subplots_adjust(top=0.98, bottom=0.05, right=0.98, left=0.1, hspace=0, wspace=0)
self.ax0 = self.fig.add_subplot(self.gs[0])
self.ax0.grid(True)
self.ax0.xaxis.set_major_formatter(ConfigParams.FORMATTER)
self.ax0.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE)
self.ax1 = self.fig.add_subplot(self.gs[1], sharex=self.ax0)
self.ax1.grid(True)
self.ax1.xaxis.set_major_formatter(ConfigParams.FORMATTER)
self.__resetAllButton__()
self.ui.doubleSpinBox_bandPassLow.setValue(Config["Filter"]["BandPassLow"])
self.ui.doubleSpinBox_bandPassHigh.setValue(Config["Filter"]["BandPassHigh"])
self.ui.spinBox_peaksValue.setValue(Config["PeaksValue"])
self.ui.pushButton_input.clicked.connect(self.__slot_btn_input__)
self.ui.pushButton_input_setting.clicked.connect(self.setting.show)
self.ui.pushButton_view.clicked.connect(self.__slot_btn_view__)
self.ui.pushButton_save.clicked.connect(self.__slot_btn_save__)
self.ui.doubleSpinBox_bandPassLow.editingFinished.connect(self.__update_config__)
self.ui.doubleSpinBox_bandPassHigh.editingFinished.connect(self.__update_config__)
self.ui.spinBox_peaksValue.editingFinished.connect(self.__update_config__)
self.ui.comboBox_method.currentTextChanged.connect(self.__update_config__)
@overrides
def closeEvent(self, event):
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.SHUTTING_DOWN))
QApplication.processEvents()
# 清空画框
if self.line_data and self.point_peak:
del self.line_data
del self.point_peak
del self.line_interval
self.canvas.draw()
# 释放资源
del self.data
self.fig.clf()
plt.close(self.fig)
self.deleteLater()
collect()
self.canvas = None
event.accept()
@staticmethod
def __reset__():
ButtonState["Current"].update(ButtonState["Default"].copy())
ButtonState["Current"]["pushButton_view"] = True
def __plot__(self):
# 清空画框
if self.line_data and self.point_peak and self.line_interval and self.point_RRIV:
try:
self.line_data.remove()
self.point_peak.remove()
self.line_interval.remove()
self.point_RRIV.remove()
except ValueError:
pass
sender = self.sender()
if sender == self.ui.pushButton_view:
self.point_RRIV, = self.ax0.plot(self.data.peak[2:], self.data.RRIV,
'r.',
label=Constants.DETECT_RPEAK_PLOT_LABEL_RRIV)
self.line_data, = self.ax1.plot(self.data.processed_data,
color=Constants.PLOT_COLOR_BLUE,
label=Constants.DETECT_RPEAK_PLOT_LABEL_ECG)
self.point_peak, = self.ax1.plot(self.data.peak, self.data.processed_data[self.data.peak],
'r*',
label=Constants.DETECT_RPEAK_PLOT_LABEL_R_PEAKS)
self.line_interval, = self.ax1.plot(self.data.interval,
color=Constants.PLOT_COLOR_GREEN,
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
else:
status = False
info = Constants.DRAWING_FAILURE
self.canvas.draw()
return status, info
def __disableAllButton__(self):
# 禁用所有按钮
all_widgets = self.centralWidget().findChildren(QWidget)
# 迭代所有部件，查找按钮并禁用它们
for widget in all_widgets:
if isinstance(widget, QPushButton):
if widget.objectName() in ButtonState["Current"].keys():
widget.setEnabled(False)
def __enableAllButton__(self):
# 启用按钮
all_widgets = self.centralWidget().findChildren(QWidget)
# 迭代所有部件，查找按钮并启用它们
for widget in all_widgets:
if isinstance(widget, QPushButton):
if widget.objectName() in ButtonState["Current"].keys():
widget.setEnabled(ButtonState["Current"][widget.objectName()])
def __resetAllButton__(self):
# 启用按钮
all_widgets = self.centralWidget().findChildren(QWidget)
# 迭代所有部件，查找按钮并启用它们
for widget in all_widgets:
if isinstance(widget, QPushButton):
if widget.objectName() in ButtonState["Default"].keys():
widget.setEnabled(ButtonState["Default"][widget.objectName()])
def __update_config__(self):
Config["Filter"]["BandPassLow"] = self.ui.doubleSpinBox_bandPassLow.value()
Config["Filter"]["BandPassHigh"] = self.ui.doubleSpinBox_bandPassHigh.value()
Config["PeaksValue"] = self.ui.spinBox_peaksValue.value()
Config["DetectMethod"] = self.ui.comboBox_method.currentText()
def finish_operation(self):
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.OPERATION_FINISHED))
self.progressbar.setValue(100)
QApplication.processEvents()
self.__enableAllButton__()
def add_progressbar(self):
self.progressbar = QProgressBar()
self.progressbar.setRange(0, 100)
self.progressbar.setValue(0)
self.progressbar.setStyleSheet(Constants.PROGRESSBAR_STYLE)
self.ui.statusbar.addPermanentWidget(self.progressbar)
def statusbar_show_msg(self, msg):
self.ui.statusbar.showMessage(msg)
def statusbar_clear_msg(self):
self.ui.statusbar.clearMessage()
def __slot_btn_input__(self):
self.__disableAllButton__()
# 清空画框
if self.line_data and self.point_peak and self.line_interval:
try:
self.line_data.remove()
self.point_peak.remove()
self.line_interval.remove()
except ValueError:
pass
self.canvas.draw()
# 清空方法列表
self.ui.comboBox_method.clear()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/2)" + Constants.DETECT_RPEAK_LOADING_METHOD))
self.progressbar.setValue(0)
QApplication.processEvents()
# 寻找方法
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"]
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_INFO)
self.update_ui_comboBox_method(method_list)
self.statusbar_show_msg(PublicFunc.format_status_msg("(2/2)" + Constants.INPUTTING_DATA))
self.progressbar.setValue(10)
QApplication.processEvents()
# 导入数据
self.data = Data()
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_INFO)
MainWindow_detect_Rpeak.__reset__()
self.finish_operation()
def __slot_btn_view__(self):
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/3)" + Constants.DETECT_RPEAK_PROCESSING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 数据预处理
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/3)" + info, Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg("(2/3)" + Constants.DETECT_RPEAK_PREDICTING_PEAK))
self.progressbar.setValue(10)
QApplication.processEvents()
# 预测峰值
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(2/3)" + 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 +
str((len(self.data.raw_data) / Config["InputConfig"]["Freq"] / 60)),
Constants.TIPS_TYPE_INFO)
PublicFunc.text_output(self.ui, Constants.DETECT_RPEAK_PEAK_AMOUNT + str(len(self.data.peak)),
Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg("(3/3)" + Constants.DRAWING_DATA))
self.progressbar.setValue(70)
QApplication.processEvents()
# 绘图
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(3/3)" + info, Constants.TIPS_TYPE_INFO)
ButtonState["Current"]["pushButton_save"] = True
self.finish_operation()
def __slot_btn_save__(self):
reply = QMessageBox.question(self, Constants.QUESTION_TITLE,
Constants.QUESTION_CONTENT + Config["Path"]["Save"],
QMessageBox.Yes | QMessageBox.No,
QMessageBox.Yes)
if reply == QMessageBox.Yes:
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/1)" + Constants.SAVING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 保存
# status, info = self.data.save()
total_rows = len(DataFrame(self.data.peak.reshape(-1)))
chunk_size = ConfigParams.DETECT_RPEAK_SAVE_CHUNK_SIZE
with open(Config["Path"]["Save"], 'w') as f:
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)
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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, info, Constants.TIPS_TYPE_INFO)
self.finish_operation()
def update_ui_comboBox_method(self, method_list):
self.ui.comboBox_method.clear()
self.ui.comboBox_method.addItems(method_list)
class Data:
def __init__(self):
self.file_path_input = Config["Path"]["Input"]
self.file_path_save = Config["Path"]["Save"]
self.raw_data = None
self.processed_data = None
self.peak = None
self.interval = None
self.RRIV = None
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"]
try:
self.raw_data = read_csv(self.file_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 True, 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"]
try:
self.processed_data = preprocess(self.raw_data,
Config["InputConfig"]["Freq"],
Config["Filter"]["BandPassLow"],
Config["Filter"]["BandPassHigh"])
except Exception:
return False, Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Filter_Exception"]
return True, 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"]
try:
self.peak, self.interval, self.RRIV = Rpeak_Detection(self.processed_data,
Config["InputConfig"]["Freq"],
Config["PeaksValue"],
Config["DetectMethod"])
except Exception:
return False, Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Predict_Exception"]
return True, 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"]
try:
# DataFrame(self.processed_data.reshape(-1)).to_csv(self.file_path_save,
# index=False,
# header=False,
# float_format='%.4f')
chunk.to_csv(self.file_path_save, mode='a', index=False, header=False)
except Exception:
return False, Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Save_Exception"]
return True, Constants.SAVING_FINISHED

9
func/Module_mainwindow.py
View File

@ -9,6 +9,7 @@ from ui.MainWindow.MainWindow_menu import Ui_Signal_Label
from func.Module_preprocess import MainWindow_preprocess
from func.Module_detect_Jpeak import MainWindow_detect_Jpeak
from func.Module_detect_Rpeak import MainWindow_detect_Rpeak
from func.utils.Constants import Constants, ConfigParams
@ -44,6 +45,7 @@ class MainWindow(QMainWindow, Ui_Signal_Label):
self.ui.pushButton_preprocess_BCG.clicked.connect(self.__slot_btn_preprocess__)
self.ui.pushButton_preprocess_ECG.clicked.connect(self.__slot_btn_preprocess__)
self.ui.pushButton_detect_Jpeak.clicked.connect(self.__slot_btn_detect_Jpeak__)
self.ui.pushButton_detect_Rpeak.clicked.connect(self.__slot_btn_detect_Rpeak__)
@staticmethod
def __read_config__():
@ -98,6 +100,13 @@ class MainWindow(QMainWindow, Ui_Signal_Label):
sampID = int(self.ui.comboBox_sampID.currentText())
self.detect_Jpeak.show(root_path, sampID)
def __slot_btn_detect_Rpeak__(self):
self.detect_Rpeak = MainWindow_detect_Rpeak()
root_path = self.ui.plainTextEdit_root_path.toPlainText()
sampID = int(self.ui.comboBox_sampID.currentText())
self.detect_Rpeak.show(root_path, sampID)
def seek_sampID(self, path):
if not Path(path).exists():

212
func/Module_preprocess.py
View File

@ -322,109 +322,6 @@ class MainWindow_preprocess(QMainWindow):
if widget.objectName() in ButtonState["Default"].keys():
widget.setEnabled(ButtonState["Default"][widget.objectName()])
def __slot_btn_input__(self):
self.__disableAllButton__()
# 清空画框
if self.line_raw_data and self.line_processed_data:
try:
self.line_raw_data.remove()
self.line_processed_data.remove()
except ValueError:
pass
self.canvas.draw()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.INPUTTING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 导入数据
self.data = Data()
status, info = self.data.open_file()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
MainWindow_preprocess.__reset__()
self.finish_operation()
def __slot_btn_view__(self):
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.PREPROCESS_PROCESSING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 数据预处理
status, info = self.data.preprocess()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.DRAWING_DATA))
self.progressbar.setValue(50)
QApplication.processEvents()
# 绘图
status, info = self.__plot__()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
ButtonState["Current"]["pushButton_save"] = True
self.finish_operation()
def __slot_btn_save__(self):
reply = QMessageBox.question(self, Constants.QUESTION_TITLE,
Constants.QUESTION_CONTENT + Config["Path"]["Save"],
QMessageBox.Yes | QMessageBox.No,
QMessageBox.Yes)
if reply == QMessageBox.Yes:
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg(Constants.SAVING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 保存
# status, info = self.data.save()
total_rows = len(DataFrame(self.data.processed_data.reshape(-1)))
chunk_size = ConfigParams.PREPROCESS_SAVE_CHUNK_SIZE
with open(Config["Path"]["Save"], 'w') as f:
for start in range(0, total_rows, chunk_size):
end = min(start + chunk_size, total_rows)
chunk = DataFrame(self.data.processed_data.reshape(-1)).iloc[start:end]
status, info = self.data.save(chunk)
progress = int((end / total_rows) * 100)
self.progressbar.setValue(progress)
QApplication.processEvents()
if not status:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, info, Constants.TIPS_TYPE_INFO)
self.finish_operation()
def __update_config__(self):
if self.mode == "BCG":
@ -450,15 +347,118 @@ class MainWindow_preprocess(QMainWindow):
self.progressbar.setRange(0, 100)
self.progressbar.setValue(0)
self.progressbar.setStyleSheet(Constants.PROGRESSBAR_STYLE)
self.ui.statusBar.addPermanentWidget(self.progressbar)
self.ui.statusbar.addPermanentWidget(self.progressbar)
def statusbar_show_msg(self, msg):
self.ui.statusBar.showMessage(msg)
self.ui.statusbar.showMessage(msg)
def statusbar_clear_msg(self):
self.ui.statusBar.clearMessage()
self.ui.statusbar.clearMessage()
def __slot_btn_input__(self):
self.__disableAllButton__()
# 清空画框
if self.line_raw_data and self.line_processed_data:
try:
self.line_raw_data.remove()
self.line_processed_data.remove()
except ValueError:
pass
self.canvas.draw()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/1)" + Constants.INPUTTING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 导入数据
self.data = Data()
status, info = self.data.open_file()
if not status:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_INFO)
MainWindow_preprocess.__reset__()
self.finish_operation()
def __slot_btn_view__(self):
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/2)" + Constants.PREPROCESS_PROCESSING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 数据预处理
status, info = self.data.preprocess()
if not status:
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/2)" + info, Constants.TIPS_TYPE_INFO)
self.statusbar_show_msg(PublicFunc.format_status_msg("(2/2)" + Constants.DRAWING_DATA))
self.progressbar.setValue(50)
QApplication.processEvents()
# 绘图
status, info = self.__plot__()
if not status:
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(2/2)" + info, Constants.TIPS_TYPE_INFO)
ButtonState["Current"]["pushButton_save"] = True
self.finish_operation()
def __slot_btn_save__(self):
reply = QMessageBox.question(self, Constants.QUESTION_TITLE,
Constants.QUESTION_CONTENT + Config["Path"]["Save"],
QMessageBox.Yes | QMessageBox.No,
QMessageBox.Yes)
if reply == QMessageBox.Yes:
self.__disableAllButton__()
self.statusbar_show_msg(PublicFunc.format_status_msg("(1/1)" + Constants.SAVING_DATA))
self.progressbar.setValue(0)
QApplication.processEvents()
# 保存
# status, info = self.data.save()
total_rows = len(DataFrame(self.data.processed_data.reshape(-1)))
chunk_size = ConfigParams.PREPROCESS_SAVE_CHUNK_SIZE
with open(Config["Path"]["Save"], 'w') as f:
for start in range(0, total_rows, chunk_size):
end = min(start + chunk_size, total_rows)
chunk = DataFrame(self.data.processed_data.reshape(-1)).iloc[start:end]
status, info = 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)
self.finish_operation()
return
else:
PublicFunc.text_output(self.ui, "(1/1)" + info, Constants.TIPS_TYPE_INFO)
PublicFunc.msgbox_output(self, info, Constants.TIPS_TYPE_INFO)
self.finish_operation()
class Data:

88
func/utils/ConfigParams.py
View File

@ -18,6 +18,8 @@ class ConfigParams:
}
}
UTF8_ENCODING: str = "utf-8"
# 目前用到这个编码的地方：
# <BCG的质量评估打标>里的保存和读取csv文件的地方（注意的是，读取原始数据时依然使用UTF-8）
GBK_ENCODING: str = "gbk"
ENDSWITH_TXT: str = ".txt"
ENDSWITH_CSV: str = ".csv"
@ -71,28 +73,41 @@ class ConfigParams:
DETECT_JPEAK_SAVE_FILENAME: str = "JPeak_revise"
DETECT_JPEAK_SAVE_CHUNK_SIZE: int = 100
# ECG的R峰算法定位
DETECT_RPEAK_CONFIG_FILE_PATH: str = "./config/Config_detect_Rpeak.yaml"
DETECT_RPEAK_CONFIG_NEW_CONTENT = {
"InputConfig": {
"Freq": 1000
},
"Filter": {
"BandPassLow": 2,
"BandPassHigh": 15
},
"PeaksValue": 200,
"DetectMethod": ""
}
DETECT_RPEAK_INPUT_ECG_FILENAME: str = "ECG_filter_"
DETECT_RPEAK_SAVE_FILENAME: str = "final_Rpeak"
DETECT_RPEAK_SAVE_CHUNK_SIZE: int = 100
# 人工纠正
# 体动标注
# TODO：弃用
# 通用
# 目前用到这个编码的地方：
# <BCG的质量评估打标>里的保存和读取csv文件的地方（注意的是，读取原始数据时依然使用UTF-8）
VALIDATOR_INTEGER = QIntValidator(-2**31, 2**31 - 1)
VALIDATOR_DOUBLE = QDoubleValidator(-1e100, 1e100, 10)
FONT: str = "Microsoft YaHei UI"
# 菜单界面
MATPLOTLIB_PLOT_PRECISION_PARAM: int = 10000
# 数据粗同步
APPROXIMATELY_ALIGN_INPUT_ORGBCG_FILENAME: str = "orgBcg_Raw_"
APPROXIMATELY_ALIGN_INPUT_PSG_FILENAME: str = "A"
@ -108,67 +123,6 @@ class ConfigParams:
APPROXIMATELY_ALIGN_BUTTERHIGHPASSFREQ_DEFAULT: float = 0.70
APPROXIMATELY_ALIGN_APPLYFREQ_DEFAULT: float = 5
# 预处理
# PREPROCESS_INPUT_BCG_FILENAME: str = "orgBcg_Raw_"
# PREPROCESS_INPUT_ECG_FILENAME: str = "ECG I_"
# PREPROCESS_SAVE_BCG_FILENAME: str = "DSbcg_sig_"
# PREPROCESS_SAVE_ECG_FILENAME: str = "ECG_filter_"
# PREPROCESS_INPUT_BCG_DEFAULT_FS: int = 1000
# PREPROCESS_INPUT_BCG_SAVE_DEFAULT_FS: int = 1000
# PREPROCESS_INPUT_ECG_DEFAULT_FS: int = 1000
# PREPROCESS_INPUT_ECG_SAVE_DEFAULT_FS: int = 1000
#
# PREPROCESS_BANDPASS_LOW_DEFAULT: int = 2
# PREPROCESS_BANDPASS_HIGH_DEFAULT: int = 10
# PREPROCESS_FILTER_ORDER_DEFAULT: int = 4
#
# PREPROCESS_FILTER_BCG: str = "bandpass"
# PREPROCESS_FILTER_ECG: str = "bandpass"
# ECG的R峰算法定位
DETECT_R_PEAK_INPUT_ECG_FILENAME: str = "ECG_filter_"
DETECT_R_PEAK_SAVE_RPEAK_FILENAME: str = "final_Rpeak"
DETECT_R_PEAK_INPUT_ECG_DEFAULT_FS: int = 1000
DETECT_R_PEAK_PEAKS_VALUE_DEFAULT: int = 200
DETECT_R_PEAK_BANDPASS_LOW_DEFAULT: int = 2
DETECT_R_PEAK_BANDPASS_HIGH_DEFAULT: int = 15
DETECT_R_PEAK_DETECT_METHOD_PT: str = "pt"
DETECT_R_PEAK_DETECT_METHOD_TA: str = "ta"
DETECT_R_PEAK_DETECT_METHOD_WT: str = "Wt"
DETECT_R_PEAK_DETECT_METHOD_HAMILTON: str = "Hamilton"
DETECT_R_PEAK_DETECT_METHOD_ENGZEE: str = "Engzee"
# BCG的J峰算法定位
DETECT_J_PEAK_INPUT_BCG_FILENAME: str = "DSbcg_sig_"
DETECT_J_PEAK_SAVE_JPEAK_FILENAME: str = "JPeak_revise"
DETECT_J_PEAK_INPUT_BCG_DEFAULT_FS: int = 1000
DETECT_J_PEAK_BANDPASS_LOW_DEFAULT: int = 2
DETECT_J_PEAK_BANDPASS_HIGH_DEFAULT: int = 10
DETECT_J_PEAK_PEAKS_VALUE_DEFAULT: int = 100
DETECT_J_PEAK_AMP_VALUE_DEFAULT: int = 5
DETECT_J_PEAK_INTERVAL_LOW_DEFAULT: int = 50
DETECT_J_PEAK_INTERVAL_HIGH_DEFAULT: int = 140
DETECT_J_PEAK_UNET_MODEL1_PKL_PATH: str = "./func/result/Fivelayer_Unet/1.pkl"
DETECT_J_PEAK_UNET_MODEL2_PKL_PATH: str = "./func/result/Fivelayer_Unet/2.pkl"
DETECT_J_PEAK_LSTMUNET_MODEL1_PKL_PATH: str = "./func/result/Fivelayer_Lstm_Unet/1.pkl"
DETECT_J_PEAK_LSTMUNET_MODEL2_PKL_PATH: str = "./func/result/Fivelayer_Lstm_Unet/2.pkl"
DETECT_J_PEAK_UNET_MODEL1_NAME: str = "Fivelayer_Unet_1"
DETECT_J_PEAK_UNET_MODEL2_NAME: str = "Fivelayer_Unet_2"
DETECT_J_PEAK_LSTMUNET_MODEL1_NAME: str = "Fivelayer_Lstm_Unet_1"
DETECT_J_PEAK_LSTMUNET_MODEL2_NAME: str = "Fivelayer_Lstm_Unet_2"
# 人工纠正
LABEL_CHECK_INPUT_BCG_FILENAME: str = "DSbcg_sig_"
LABEL_CHECK_INPUT_JPEAK_FILENAME: str = "JPeak_revise"

79
func/utils/Constants.py
View File

@ -15,10 +15,6 @@ class Constants:
INPUT_FINISHED: str = "导入完成"
INPUT_FAILURE: str = "导入失败"
LOADING_MODEL: str = "正在读取模型"
LOAD_FINISHED: str = "读取完成"
LOAD_FAILURE: str = "读取失败"
DRAWING_DATA: str = "正在绘制图形"
DRAWING_FINISHED: str = "绘制完成"
DRAWING_FAILURE: str = "绘制失败"
@ -31,7 +27,7 @@ class Constants:
OPERATION_FAILURE: str = "操作失败"
UNKNOWN_ERROR: str = "未知错误"
SHUTTING_DOWN: str = "正在关闭窗口"
SHUTTING_DOWN: str = "正在释放内存资源"
QUESTION_TITLE: str = "警告：确认操作"
QUESTION_CONTENT: str = "你确定要保存结果到"
@ -89,6 +85,10 @@ class Constants:
PREPROCESS_OUTPUT_INPUT_AMP_OFFSET: int = 1850
# BCG的J峰算法定位
DETECT_JPEAK_LOADING_MODEL: str = "正在读取模型"
DETECT_JPEAK_LOAD_FINISHED: str = "读取完成"
DETECT_JPEAK_LOAD_FAILURE: str = "读取失败"
DETECT_JPEAK_PROCESSING_DATA: str = "正在处理数据"
DETECT_JPEAK_PROCESS_FINISHED: str = "处理完成"
DETECT_JPEAK_PROCESS_FAILURE: str = "处理失败"
@ -103,7 +103,7 @@ class Constants:
"Model_Path_Not_Exist": "（模型路径不存在）",
"Model_File_Not_Exist": "（模型文件不存在）",
"Read_Model_Exception": "（读取模型异常）",
"Predict_Exception": "（模型预测异常）",
"Predict_Exception": "（峰值预测异常）",
"Raw_Data_Not_Exist": "（原始数据不存在）",
"Filter_Exception": "（滤波器异常）",
"Processed_Data_Not_Exist": "（处理后数据不存在）",
@ -111,12 +111,50 @@ class Constants:
"Save_Exception": "（保存异常）"
}
DETECT_J_PEAK_DATA_LENGTH_POINTS: str = "数据长度（点数）："
DETECT_J_PEAK_DURATION_MIN: str = "数据时长（分钟）："
DETECT_J_PEAK_JPEAK_AMOUNT: str = "J峰个数："
DETECT_J_PEAK_PLOT_LABEL_BCG: str = "BCG_Processed"
DETECT_J_PEAK_PLOT_LABEL_J_PEAKS: str = "J_Peaks"
DETECT_J_PEAK_PLOT_LABEL_INTERVAL: str = "Interval"
DETECT_JPEAK_DATA_LENGTH_POINTS: str = "数据长度（点数）："
DETECT_JPEAK_DURATION_MIN: str = "数据时长（分钟）："
DETECT_JPEAK_PEAK_AMOUNT: str = "J峰个数："
DETECT_JPEAK_PLOT_LABEL_BCG: str = "BCG_Processed"
DETECT_JPEAK_PLOT_LABEL_J_PEAKS: str = "J_Peaks"
DETECT_JPEAK_PLOT_LABEL_INTERVAL: str = "Interval"
# ECG的R峰算法定位
DETECT_RPEAK_LOADING_METHOD: str = "正在读取方法"
DETECT_RPEAK_LOAD_FINISHED: str = "读取完成"
DETECT_RPEAK_LOAD_FAILURE: str = "读取失败"
DETECT_RPEAK_PROCESSING_DATA: str = "正在处理数据"
DETECT_RPEAK_PROCESS_FINISHED: str = "处理完成"
DETECT_RPEAK_PROCESS_FAILURE: str = "处理失败"
DETECT_RPEAK_PREDICTING_PEAK: str = "正在预测峰值"
DETECT_RPEAK_PREDICT_FINISHED: str = "预测完成"
DETECT_RPEAK_PREDICT_FAILURE: str = "预测失败"
DETECT_RPEAK_FAILURE_REASON = {
"Data_Path_Not_Exist": "（数据路径不存在）",
"Read_Data_Exception": "（读取数据异常）",
"Method_Not_Exist": "（检测方法不存在）",
"Read_Method_Exception": "（读取方法异常）",
"Predict_Exception": "（峰值预测异常）",
"Raw_Data_Not_Exist": "（原始数据不存在）",
"Filter_Exception": "（滤波器异常）",
"Processed_Data_Not_Exist": "（处理后数据不存在）",
"Peak_Not_Exist": "（预测的峰值不存在）",
"Save_Exception": "（保存异常）"
}
DETECT_RPEAK_DATA_LENGTH_POINTS: str = "数据长度（点数）："
DETECT_RPEAK_DURATION_MIN: str = "数据时长（分钟）："
DETECT_RPEAK_PEAK_AMOUNT: str = "R峰个数："
DETECT_RPEAK_PLOT_LABEL_RRIV: str = "RRIV"
DETECT_RPEAK_PLOT_LABEL_ECG: str = "ECG"
DETECT_RPEAK_PLOT_LABEL_R_PEAKS: str = "R_Peaks"
DETECT_RPEAK_PLOT_LABEL_INTERVAL: str = "Interval"
# 人工纠正
# 体动标注
# TODO：弃用
@ -124,7 +162,6 @@ class Constants:
FOLDER_DIR_NOT_EXIST_THEN_CREATE: str = "检测到保存路径所指向的文件夹不存在，已创建相应文件夹"
# 菜单界面
MAINWINDOW_ROOT_PATH_NOT_EXIST: str = "根目录路径输入错误"
MAINWINDOW_MSGBOX_TITLE: str = "消息"
@ -132,7 +169,6 @@ class Constants:
MAINWINDOW_BACK_TO_MENU: str = "返回主菜单"
MAINWINDOW_QUESTION_BACK_TO_MENU: str = "确定要返回主菜单吗"
# 数据粗同步
APPROXIMATELY_ALIGN_FILES_NOT_FOUND: str = f"无法找到{ConfigParams.APPROXIMATELY_ALIGN_INPUT_ORGBCG_FILENAME}{ConfigParams.ENDSWITH_TXT}或{ConfigParams.APPROXIMATELY_ALIGN_INPUT_PSG_FILENAME}{ConfigParams.ENDSWITH_EDF}，无法执行<数据粗同步>"
APPROXIMATELY_ALIGN_FILES_FOUND: str = f"找到{ConfigParams.APPROXIMATELY_ALIGN_INPUT_ORGBCG_FILENAME}{ConfigParams.ENDSWITH_TXT}和{ConfigParams.APPROXIMATELY_ALIGN_INPUT_PSG_FILENAME}{ConfigParams.ENDSWITH_EDF}"
@ -140,21 +176,6 @@ class Constants:
APPROXIMATELY_ALIGN_RUNNING: str = "开始执行任务<数据粗同步>"
APPROXIMATELY_RECORD_NOT_FOUND: str = "没有保存记录"
# ECG的R峰算法定位
DETECT_R_PEAK_FILES_NOT_FOUND: str = f"无法找到{ConfigParams.DETECT_R_PEAK_INPUT_ECG_FILENAME}{ConfigParams.ENDSWITH_TXT}，无法执行<R峰提取>"
DETECT_R_PEAK_FILES_FOUND: str = f"找到{ConfigParams.DETECT_R_PEAK_INPUT_ECG_FILENAME}{ConfigParams.ENDSWITH_TXT}"
DETECT_R_PEAK_RUNNING: str = "开始执行任务<ECG的R峰算法定位>"
DETECT_R_PEAK_PLOT_LABEL_RRIV: str = "RRIV"
DETECT_R_PEAK_PLOT_LABEL_ECG: str = "ECG"
DETECT_R_PEAK_PLOT_LABEL_R_PEAKS: str = "R_peaks"
DETECT_R_PEAK_PLOT_LABEL_INTERVAL: str = "Interval"
DETECT_R_PEAK_DATA_LENGTH_POINTS: str = "数据长度（点数）："
DETECT_R_PEAK_DURATION_MIN: str = "数据时长（分钟）："
DETECT_R_PEAK_RPEAK_AMOUNT: str = "R峰个数："
# 人工纠正
LABEL_CHECK_FILES_BCG_NOT_FOUND: str = f"无法找到{ConfigParams.LABEL_CHECK_INPUT_BCG_FILENAME}{ConfigParams.ENDSWITH_TXT}或{ConfigParams.LABEL_CHECK_INPUT_JPEAK_FILENAME}{ConfigParams.ENDSWITH_TXT}，无法执行<BCG的J峰人工纠正>"
LABEL_CHECK_FILES_BCG_FOUND: str = f"找到{ConfigParams.LABEL_CHECK_INPUT_BCG_FILENAME}{ConfigParams.ENDSWITH_TXT}和{ConfigParams.LABEL_CHECK_INPUT_JPEAK_FILENAME}{ConfigParams.ENDSWITH_TXT}"

49
func/utils/detect_Rpeak.py
View File

@ -29,31 +29,38 @@ def find_TPeak(data,peaks,th=50):
return_peak.append(argmax(data[min_win:max_win])+min_win)
return array(return_peak)
def Rpeak_Detection(raw_ecg,fs,low_cut,high_cut,th1,detector_method):
detectors = Detectors(sampling_frequency=fs)
method_dic = {'pt': detectors.pan_tompkins_detector,
'ta': detectors.two_average_detector,
"Engzee": detectors.engzee_detector,
"Wt": detectors.swt_detector,
"Christov": detectors.christov_detector,
"Hamilton": detectors.hamilton_detector
}
detectormethods = method_dic[detector_method]
def preprocess(raw_ecg, fs, low_cut, high_cut):
# raw_ecg = raw_ecg[200*sample_rate:]
preprocessing = BCG_Operation(sample_rate=fs) # 对ECG做了降采样处理
raw_ecg = preprocessing.Butterworth(raw_ecg, "bandpass", low_cut=low_cut, high_cut=high_cut, order=3) * 4
#######################限制幅值处理############################################
# for i in range(len(raw_ecg)):
# if raw_ecg[i] > 300 or raw_ecg[i] < -300:
# raw_ecg[i] = 0
##############################################################################
return raw_ecg
R_peak = array(detectormethods(raw_ecg)) - 100
# R_peak = np.array(detectors.pan_tompkins_detector(raw_ecg))-100
# 界面会通过这个函数获取方法列表，此函数的返回值务必和Rpeak_Detection()中的方法名称对应否则程序或许直接崩溃
def get_method():
return ["pt", "ta", "Engzee", "Wt", "Christov", "Hamilton"]
R_peak = find_TPeak(raw_ecg, R_peak, th=int(th1 * fs / 1000))
R_peak = refinement(raw_ecg, R_peak)
def Rpeak_Detection(ecg,fs,th1,detector_method):
detectors = Detectors(sampling_frequency=fs)
if detector_method == "pt":
detectormethods = detectors.pan_tompkins_detector
elif detector_method == "ta":
detectormethods = detectors.two_average_detector
elif detector_method == "Engzee":
detectormethods = detectors.engzee_detector
elif detector_method == "Wt":
detectormethods = detectors.swt_detector
elif detector_method == "Christov":
detectormethods = detectors.christov_detector
elif detector_method == "Hamilton":
detectormethods = detectors.hamilton_detector
else:
raise Exception
R_peak = array(detectormethods(ecg)) - 100
R_peak = find_TPeak(ecg, R_peak, th=int(th1 * fs / 1000))
R_peak = refinement(ecg, R_peak)
RR_Interval = full(len(R_peak) - 1, nan)
@ -64,7 +71,7 @@ def Rpeak_Detection(raw_ecg,fs,low_cut,high_cut,th1,detector_method):
for i in range(len(RR_Interval) - 1):
RRIV[i] = RR_Interval[i + 1] - RR_Interval[i]
Interval = full(len(raw_ecg), nan)
Interval = full(len(ecg), nan)
for i in range(len(R_peak) - 1):
Interval[R_peak[i]: R_peak[i + 1]] = R_peak[i + 1] - R_peak[i]