from gc import collect from pathlib import Path from traceback import format_exc import matplotlib.pyplot as plt from PySide6.QtCore import QCoreApplication from PySide6.QtGui import QAction, QFont, QColor from PySide6.QtWidgets import QMessageBox, QMainWindow, QApplication, QButtonGroup, QTableWidget, QTableWidgetItem, \ QHeaderView from matplotlib import gridspec from matplotlib.backends.backend_qt import NavigationToolbar2QT from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg from numpy import array, zeros, append, linspace, place, nan from overrides import overrides from pandas import read_csv, DataFrame, Series, concat from yaml import dump, load, FullLoader from func.Filters.Preprocessing import Butterworth_for_ECG_PreProcess from func.utils.PublicFunc import PublicFunc from func.utils.Constants import Constants, ConfigParams from func.utils.Result import Result from ui.MainWindow.MainWindow_SA_label import Ui_MainWindow_SA_label from ui.setting.SA_label_input_setting import Ui_MainWindow_SA_label_input_setting Config = { } ButtonState = { "Default": { "pushButton_input_setting": True, "pushButton_input": True, "pushButton_quick_remark_input_waitingForTalk": False, "pushButton_quick_remark_input_maybeDesaturation": False, "pushButton_quick_remark_input_maybeWrongLabeled": False, "pushButton_quick_remark_input_durationNoEnough": False, "pushButton_quick_remark_input_littleChange": False, "pushButton_quick_remark_input_noNormalRespBetweenArtifact": False, "pushButton_quick_remark_input_lowSignalNoiseRatio": False, "pushButton_quick_remark_input_changeOnMiddle": False, "pushButton_prev": False, "pushButton_next": False, "pushButton_confirmLabel": False, "pushButton_previous10s": False, "pushButton_previous30s": False, "pushButton_previous60s": False, "pushButton_next10s": False, "pushButton_next30s": False, "pushButton_next60s": False }, "Current": { "pushButton_input_setting": True, "pushButton_input": True, "pushButton_quick_remark_input_waitingForTalk": False, "pushButton_quick_remark_input_maybeDesaturation": False, "pushButton_quick_remark_input_maybeWrongLabeled": False, "pushButton_quick_remark_input_durationNoEnough": False, "pushButton_quick_remark_input_littleChange": False, "pushButton_quick_remark_input_noNormalRespBetweenArtifact": False, "pushButton_quick_remark_input_lowSignalNoiseRatio": False, "pushButton_quick_remark_input_changeOnMiddle": False, "pushButton_prev": False, "pushButton_next": False, "pushButton_confirmLabel": False, "pushButton_previous10s": False, "pushButton_previous30s": False, "pushButton_previous60s": False, "pushButton_next10s": False, "pushButton_next30s": False, "pushButton_next60s": False } } class SettingWindow(QMainWindow): def __init__(self, root_path, sampID): super(SettingWindow, self).__init__() self.ui = Ui_MainWindow_SA_label_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_signal_OrgBCG.valueChanged.connect(self.__update_ui__) self.ui.spinBox_input_freq_signal_Tho.valueChanged.connect(self.__update_ui__) self.ui.spinBox_input_freq_signal_Abd.valueChanged.connect(self.__update_ui__) self.ui.spinBox_input_freq_signal_FlowT.valueChanged.connect(self.__update_ui__) self.ui.spinBox_input_freq_signal_FlowP.valueChanged.connect(self.__update_ui__) self.ui.spinBox_input_freq_signal_SpO2.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.SA_LABEL_CONFIG_FILE_PATH).exists(): with open(ConfigParams.SA_LABEL_CONFIG_FILE_PATH, "w") as f: dump(ConfigParams.SA_LABEL_CONFIG_NEW_CONTENT, f) with open(ConfigParams.SA_LABEL_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_OrgBCG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED / Path(str(self.sampID)))), "Input_Tho": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Input_Abd": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Input_FlowT": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Input_FlowP": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Input_SpO2": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Input_Artifact": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_LABEL / Path(str(self.sampID)))), "Input_Label": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)))), "Save": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_LABEL / Path(str(self.sampID)))), "Save_2": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_LABEL / Path(str(self.sampID)))) }, "PlotEventIndex": 0, "TimeMoveCount": 0, "BCG_SP": 0, "BCG_EP": 0, "EventLabelIndexList": [] }) # 数据回显 self.ui.spinBox_input_freq_signal_OrgBCG.setValue(Config["InputConfig"]["OrgBCGFreq"]) self.ui.spinBox_input_freq_signal_Tho.setValue(Config["InputConfig"]["ThoFreq"]) self.ui.spinBox_input_freq_signal_Abd.setValue(Config["InputConfig"]["AbdFreq"]) self.ui.spinBox_input_freq_signal_FlowT.setValue(Config["InputConfig"]["FlowTFreq"]) self.ui.spinBox_input_freq_signal_FlowP.setValue(Config["InputConfig"]["FlowPFreq"]) self.ui.spinBox_input_freq_signal_SpO2.setValue(Config["InputConfig"]["SpO2Freq"]) self.ui.plainTextEdit_file_path_input_signal_OrgBCG.setPlainText(Config["Path"]["Input_OrgBCG"]) self.ui.plainTextEdit_file_path_input_signal_Tho.setPlainText(Config["Path"]["Input_Tho"]) self.ui.plainTextEdit_file_path_input_signal_Abd.setPlainText(Config["Path"]["Input_Abd"]) self.ui.plainTextEdit_file_path_input_signal_FlowT.setPlainText(Config["Path"]["Input_FlowT"]) self.ui.plainTextEdit_file_path_input_signal_FlowP.setPlainText(Config["Path"]["Input_FlowP"]) self.ui.plainTextEdit_file_path_input_signal_SpO2.setPlainText(Config["Path"]["Input_SpO2"]) self.ui.plainTextEdit_file_path_input_artifact.setPlainText(Config["Path"]["Input_Artifact"]) self.ui.plainTextEdit_file_path_input_label.setPlainText(Config["Path"]["Input_Label"]) self.ui.plainTextEdit_file_path_save.setPlainText(Config["Path"]["Save"]) self.ui.plainTextEdit_file_path_save_2.setPlainText(Config["Path"]["Save_2"]) def __write_config__(self): # 从界面写入配置 Config["InputConfig"]["OrgBCGFreq"] = self.ui.spinBox_input_freq_signal_OrgBCG.value() Config["InputConfig"]["ThoFreq"] = self.ui.spinBox_input_freq_signal_Tho.value() Config["InputConfig"]["AbdFreq"] = self.ui.spinBox_input_freq_signal_Abd.value() Config["InputConfig"]["FlowTFreq"] = self.ui.spinBox_input_freq_signal_FlowT.value() Config["InputConfig"]["FlowPFreq"] = self.ui.spinBox_input_freq_signal_FlowP.value() Config["InputConfig"]["SpO2Freq"] = self.ui.spinBox_input_freq_signal_SpO2.value() Config["Path"]["Input_OrgBCG"] = self.ui.plainTextEdit_file_path_input_signal_OrgBCG.toPlainText() Config["Path"]["Input_Tho"] = self.ui.plainTextEdit_file_path_input_signal_Tho.toPlainText() Config["Path"]["Input_Abd"] = self.ui.plainTextEdit_file_path_input_signal_Abd.toPlainText() Config["Path"]["Input_FlowT"] = self.ui.plainTextEdit_file_path_input_signal_FlowT.toPlainText() Config["Path"]["Input_FlowP"] = self.ui.plainTextEdit_file_path_input_signal_FlowP.toPlainText() Config["Path"]["Input_SpO2"] = self.ui.plainTextEdit_file_path_input_signal_SpO2.toPlainText() Config["Path"]["Input_Artifact"] = self.ui.plainTextEdit_file_path_input_artifact.toPlainText() Config["Path"]["Input_Label"] = self.ui.plainTextEdit_file_path_input_label.toPlainText() Config["Path"]["Save"] = self.ui.plainTextEdit_file_path_save.toPlainText() Config["Path"]["Save_2"] = self.ui.plainTextEdit_file_path_save_2.toPlainText() # 保存配置到文件 self.config["InputConfig"]["OrgBCGFreq"] = self.ui.spinBox_input_freq_signal_OrgBCG.value() self.config["InputConfig"]["ThoFreq"] = self.ui.spinBox_input_freq_signal_Tho.value() self.config["InputConfig"]["AbdFreq"] = self.ui.spinBox_input_freq_signal_Abd.value() self.config["InputConfig"]["FlowTFreq"] = self.ui.spinBox_input_freq_signal_FlowT.value() self.config["InputConfig"]["FlowPFreq"] = self.ui.spinBox_input_freq_signal_FlowP.value() self.config["InputConfig"]["SpO2Freq"] = self.ui.spinBox_input_freq_signal_SpO2.value() with open(ConfigParams.SA_LABEL_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_signal_OrgBCG.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.ORGBCG_SYNC + str(self.ui.spinBox_input_freq_signal_OrgBCG.value()) + ConfigParams.ENDSWITH_TXT)))) self.ui.plainTextEdit_file_path_input_signal_Tho.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.THO_SYNC + str(self.ui.spinBox_input_freq_signal_Tho.value()) + ConfigParams.ENDSWITH_TXT)))) self.ui.plainTextEdit_file_path_input_signal_Abd.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.ABD_SYNC + str(self.ui.spinBox_input_freq_signal_Abd.value()) + ConfigParams.ENDSWITH_TXT)))) self.ui.plainTextEdit_file_path_input_signal_FlowT.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.FLOWT_SYNC + str(self.ui.spinBox_input_freq_signal_FlowT.value()) + ConfigParams.ENDSWITH_TXT)))) self.ui.plainTextEdit_file_path_input_signal_FlowP.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.FLOWP_SYNC + str(self.ui.spinBox_input_freq_signal_FlowP.value()) + ConfigParams.ENDSWITH_TXT)))) self.ui.plainTextEdit_file_path_input_signal_SpO2.setPlainText( str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED / Path(str(self.sampID)) / Path(ConfigParams.SPO2_SYNC + str(self.ui.spinBox_input_freq_signal_SpO2.value()) + ConfigParams.ENDSWITH_TXT)))) class MainWindow_SA_label(QMainWindow): def __init__(self): super(MainWindow_SA_label, self).__init__() self.ui = Ui_MainWindow_SA_label() self.ui.setupUi(self) self.root_path = None self.sampID = None self.data = None self.setting = None self.buttonGroup_event = None self.buttonGroup_class = None # 初始化进度条 self.progressbar = None PublicFunc.add_progressbar(self) # 初始化画框 self.fig = None self.canvas = None self.figToolbar = None self.gs = None # 设定事件和其对应颜色 # event_code color event # 0 黑色 背景 # 1 粉色 低通气 # 2 蓝色 中枢性 # 3 红色 阻塞型 # 4 灰色 混合型 # 5 绿色 血氧饱和度下降 # 6 橙色 大体动 # 7 橙色 小体动 # 8 橙色 深呼吸 # 9 橙色 脉冲体动 # 10 橙色 无效片段 self.color_cycle = ["black", "pink", "blue", "red", "silver", "green", "orange", "orange", "orange", "orange", "orange"] 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 = CustomNavigationToolbar(self.canvas, self) self.figToolbar.action_Reset_Signal_and_Time.setEnabled(False) for action in self.figToolbar._actions.values(): action.setEnabled(False) for action in self.figToolbar.actions(): if action.text() == "Home" or action.text() == "Subplots" or action.text() == "Customize": self.figToolbar.removeAction(action) self.figToolbar.action_Reset_Signal_and_Time.triggered.connect(self.toggle_resetOriginalView) self.ui.verticalLayout_canvas.addWidget(self.canvas) self.ui.verticalLayout_canvas.addWidget(self.figToolbar) self.gs = gridspec.GridSpec(7, 1, height_ratios=[1, 1, 1, 1, 1, 3, 2]) 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.ax1.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) self.ax2 = self.fig.add_subplot(self.gs[2], sharex=self.ax0) self.ax2.grid(True) self.ax2.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax2.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) self.ax3 = self.fig.add_subplot(self.gs[3], sharex=self.ax0) self.ax3.grid(True) self.ax3.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax3.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) self.ax4 = self.fig.add_subplot(self.gs[4], sharex=self.ax0) self.ax4.grid(True) self.ax4.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax4.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) self.ax5 = self.fig.add_subplot(self.gs[5], sharex=self.ax0) self.ax5.grid(True) self.ax5.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax5.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) self.ax6 = self.fig.add_subplot(self.gs[6], sharex=self.ax0) self.ax6.grid(True) self.ax6.xaxis.set_major_formatter(ConfigParams.FORMATTER) PublicFunc.__resetAllButton__(self, ButtonState) self.ui.tableWidget_label.setHorizontalHeaderLabels(['原事件类型', '事件类型', '标签类型', '起始时间(s)', '终止时间(s)']) self.ui.tableWidget_label_add.setHorizontalHeaderLabels(['事件类型', '标签类型', '起始时间(s)', '终止时间(s)']) self.ui.tableWidget_label.setEditTriggers(QTableWidget.NoEditTriggers) self.ui.tableWidget_label_add.setEditTriggers(QTableWidget.NoEditTriggers) self.ui.tableWidget_label.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.ui.tableWidget_label_add.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.buttonGroup_event = QButtonGroup(self) self.buttonGroup_class = QButtonGroup(self) self.buttonGroup_event.addButton(self.ui.radioButton_OSA) self.buttonGroup_event.addButton(self.ui.radioButton_CSA) self.buttonGroup_event.addButton(self.ui.radioButton_MSA) self.buttonGroup_event.addButton(self.ui.radioButton_HPY) self.buttonGroup_class.addButton(self.ui.radioButton_1_class) self.buttonGroup_class.addButton(self.ui.radioButton_2_class) self.buttonGroup_class.addButton(self.ui.radioButton_3_class) self.ui.radioButton_OSA.setChecked(True) self.ui.radioButton_2_class.setChecked(True) self.ui.pushButton_input.clicked.connect(self.__slot_btn_input__) self.ui.pushButton_input_setting.clicked.connect(self.setting.show) self.ui.tableWidget_label.cellDoubleClicked.connect(self.__slot_tableWidget_label_on_cell_double_clicked__) self.ui.lineEdit_filter_label.textChanged.connect(self.__slot_lineEdit_filter_label__) self.ui.lineEdit_filter_label_add.textChanged.connect(self.__slot_lineEdit_filter_label__) self.ui.pushButton_prev.clicked.connect(self.__slot_btn_move__) self.ui.pushButton_next.clicked.connect(self.__slot_btn_move__) self.ui.pushButton_confirmLabel.clicked.connect(self.__slot_btn_confirmLabel__) self.ui.pushButton_quick_remark_input_waitingForTalk.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_maybeDesaturation.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_maybeWrongLabeled.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_durationNoEnough.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_littleChange.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_noNormalRespBetweenArtifact.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_lowSignalNoiseRatio.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_quick_remark_input_changeOnMiddle.clicked.connect(self.__slot_btn_quick_remark__) self.ui.pushButton_previous10s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.pushButton_previous30s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.pushButton_previous60s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.pushButton_next10s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.pushButton_next30s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.pushButton_next60s.clicked.connect(self.__slot_btn_moveBySecond__) self.ui.checkBox_examineBySecond.clicked.connect(self.__slot_checkBox_examineBySecond__) self.ui.checkBox_examineLabeled.clicked.connect(self.__slot_checkBox_examineLabeled__) self.ui.pushButton_prev.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_BTN_PREV_SHORTCUT_KEY)) self.ui.pushButton_confirmLabel.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_BTN_CONFIRMLABEL_SHORTCUT_KEY)) self.ui.pushButton_next.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_BTN_NEXT_SHORTCUT_KEY)) self.ui.radioButton_OSA.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_OSA_SHORTCUT_KEY)) self.ui.radioButton_CSA.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_CSA_SHORTCUT_KEY)) self.ui.radioButton_MSA.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_MSA_SHORTCUT_KEY)) self.ui.radioButton_HPY.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_HPY_SHORTCUT_KEY)) self.ui.radioButton_1_class.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_1_CLASS_SHORTCUT_KEY)) self.ui.radioButton_2_class.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_2_CLASS_SHORTCUT_KEY)) self.ui.radioButton_3_class.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_RADIOBUTTON_3_CLASS_SHORTCUT_KEY)) self.ui.pushButton_quick_remark_input_waitingForTalk.setShortcut(QCoreApplication.translate("MainWindow", ConfigParams.SA_LABEL_BTN_QUICK_REMARK_WAITINGFORTALK_SHORTCUT_KEY)) @overrides def closeEvent(self, event): 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() # 清空画框 if self.ax0 is not None: self.ax0.clear() if self.ax1 is not None: self.ax1.clear() if self.ax2 is not None: self.ax2.clear() if self.ax3 is not None: self.ax3.clear() if self.ax4 is not None: self.ax4.clear() if self.ax5 is not None: self.ax5.clear() if self.ax6 is not None: self.ax6.clear() # 释放资源 del self.data self.fig.clf() plt.close(self.fig) self.deleteLater() collect() self.canvas = None event.accept() else: event.ignore() def __reset__(self): ButtonState["Current"].update(ButtonState["Default"].copy()) def __plot__(self, *args, **kwargs): # 清空画框 self.reset_axes() sender = self.sender() if sender == self.ui.pushButton_input: try: self.draw_one_event() self.canvas.draw() except Exception as e: return Result().failure(info=Constants.DRAW_FAILURE + "\n" + format_exc()) elif sender == self.ui.tableWidget_label or sender == self.ui.tableWidget_label_add: try: self.draw_one_event() self.canvas.draw() except Exception as e: return Result().failure(info=Constants.DRAW_FAILURE + "\n" + format_exc()) elif sender == self.ui.pushButton_prev or sender == self.ui.pushButton_next: try: self.draw_one_event() self.canvas.draw() except Exception as e: return Result().failure(info=Constants.DRAW_FAILURE + "\n" + format_exc()) elif (sender == self.ui.pushButton_previous10s or sender == self.ui.pushButton_previous30s or sender == self.ui.pushButton_previous60s or sender == self.ui.pushButton_next10s or sender == self.ui.pushButton_next30s or sender == self.ui.pushButton_next60s): try: self.draw_new_event(*args, **kwargs) self.canvas.draw() except Exception as e: return Result().failure(info=Constants.DRAW_FAILURE + "\n" + format_exc()) return Result().success(info=Constants.DRAW_FINISHED) def update_UI_Args(self): try: if self.data is not None and self.data.df_corrected is not None: if not str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "remark"]) == "nan": self.ui.lineEdit_remark.setText( str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "remark"])) else: self.ui.lineEdit_remark.setText("") if not (self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "isLabeled"] == -1): if self.data.df_corrected.at[Config["PlotEventIndex"], "correct_EventsType"] == "Obstructive apnea": self.ui.radioButton_OSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "correct_EventsType"] == "Central apnea": self.ui.radioButton_CSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "correct_EventsType"] == "Mixed apnea": self.ui.radioButton_MSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "correct_EventsType"] == "Hypopnea": self.ui.radioButton_HPY.setChecked(True) else: if self.data.df_corrected.at[Config["PlotEventIndex"], "Event type"] == "Obstructive apnea": self.ui.radioButton_OSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "Event type"] == "Central apnea": self.ui.radioButton_CSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "Event type"] == "Mixed apnea": self.ui.radioButton_MSA.setChecked(True) elif self.data.df_corrected.at[Config["PlotEventIndex"], "Event type"] == "Hypopnea": self.ui.radioButton_HPY.setChecked(True) if str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"]) == "1": self.ui.radioButton_1_class.setChecked(True) elif str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"]) == "2": self.ui.radioButton_2_class.setChecked(True) elif str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"]) == "3": self.ui.radioButton_3_class.setChecked(True) else: self.ui.radioButton_2_class.setChecked(True) except Exception as e: return Result().failure(info=Constants.UPDATE_FAILURE + Constants.FAILURE_REASON["Update_Info_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.UPDATE_FINISHED) def update_tableWidget(self): try: if self.data.df_corrected is not None: self.ui.tableWidget_label.setRowCount(len(self.data.df_corrected)) for index, row in self.data.df_corrected.iterrows(): Event_type = row.get("Event type", None) if Event_type == "Obstructive apnea": item = QTableWidgetItem("OSA_org") elif Event_type == "Central apnea": item = QTableWidgetItem("CSA_org") elif Event_type == "Mixed apnea": item = QTableWidgetItem("MSA_org") elif Event_type == "Hypopnea": item = QTableWidgetItem("HPY_org") else: item = QTableWidgetItem("None") self.ui.tableWidget_label.setItem(index, 0, item) correct_EventsType = row.get("correct_EventsType", None) if correct_EventsType == "Obstructive apnea": item = QTableWidgetItem("OSA") elif correct_EventsType == "Central apnea": item = QTableWidgetItem("CSA") elif correct_EventsType == "Mixed apnea": item = QTableWidgetItem("MSA") elif correct_EventsType == "Hypopnea": item = QTableWidgetItem("HPY") else: item = QTableWidgetItem("None") self.ui.tableWidget_label.setItem(index, 1, item) score = row.get("score", None) if score == 1: item = QTableWidgetItem("一类") elif score == 2: item = QTableWidgetItem("二类") elif score == 3: item = QTableWidgetItem("删除") else: item = QTableWidgetItem("None") self.ui.tableWidget_label.setItem(index, 2, item) correct_Start = row.get("correct_Start", None) if correct_Start < 0 or correct_Start * Config["InputConfig"]["OrgBCGFreq"] > len(self.data.OrgBCG): item = QTableWidgetItem("None") else: item = QTableWidgetItem(str(correct_Start)) self.ui.tableWidget_label.setItem(index, 3, item) correct_End = row.get("correct_End", None) if correct_End < 0 or correct_End * Config["InputConfig"]["OrgBCGFreq"] > len(self.data.OrgBCG): item = QTableWidgetItem("None") else: item = QTableWidgetItem(str(correct_End)) self.ui.tableWidget_label.setItem(index, 4, item) remark = row.get("remark", None) if str(remark) != "" and str(remark) != "nan" and row.get("isLabeled", None) == 1: for col in range(self.ui.tableWidget_label.columnCount()): item = self.ui.tableWidget_label.item(index, col) item.setBackground(QColor(255, 200, 200)) else: for col in range(self.ui.tableWidget_label.columnCount()): item = self.ui.tableWidget_label.item(index, col) item.setBackground(QColor(255, 255, 255)) if self.data.df_addNew is not None: self.ui.tableWidget_label_add.setRowCount(len(self.data.df_addNew)) for index, row in self.data.df_addNew.iterrows(): correct_EventsType = row.get("correct_EventsType", None) if correct_EventsType == "Obstructive apnea": item = QTableWidgetItem("OSA") elif correct_EventsType == "Central apnea": item = QTableWidgetItem("CSA") elif correct_EventsType == "Mixed apnea": item = QTableWidgetItem("MSA") elif correct_EventsType == "Hypopnea": item = QTableWidgetItem("HPY") else: item = QTableWidgetItem("None") self.ui.tableWidget_label_add.setItem(index, 0, item) score = row.get("score", None) if score == 1: item = QTableWidgetItem("一类") elif score == 2: item = QTableWidgetItem("二类") elif score == 3: item = QTableWidgetItem("删除") else: item = QTableWidgetItem("None") self.ui.tableWidget_label_add.setItem(index, 1, item) correct_Start = row.get("correct_Start", None) if correct_Start < 0 or correct_Start * Config["InputConfig"]["OrgBCGFreq"] > len(self.data.OrgBCG): item = QTableWidgetItem("None") else: item = QTableWidgetItem(str(correct_Start)) self.ui.tableWidget_label_add.setItem(index, 2, item) correct_End = row.get("correct_End", None) if correct_End < 0 or correct_End * Config["InputConfig"]["OrgBCGFreq"] > len(self.data.OrgBCG): item = QTableWidgetItem("None") else: item = QTableWidgetItem(str(correct_End)) self.ui.tableWidget_label_add.setItem(index, 3, item) remark = row.get("remark", None) if str(remark) != "" and str(remark) != "nan" and row.get("isLabeled", None) == 1: for col in range(self.ui.tableWidget_label_add.columnCount()): item = self.ui.tableWidget_label_add.item(index, col) item.setBackground(QColor(255, 200, 200)) else: for col in range(self.ui.tableWidget_label_add.columnCount()): item = self.ui.tableWidget_label_add.item(index, col) item.setBackground(QColor(255, 255, 255)) except Exception as e: return Result().failure(info=Constants.UPDATE_FAILURE + Constants.FAILURE_REASON["Update_tableWidget_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.UPDATE_FINISHED) def __slot_btn_input__(self): PublicFunc.__disableAllButton__(self, ButtonState) self.reset_axes() self.canvas.draw() self.data = Data() # 导入数据 PublicFunc.progressbar_update(self, 1, 7, Constants.INPUTTING_DATA, 0) result = self.data.open_file() if not result.status: PublicFunc.text_output(self.ui, "(1/7)" + 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/7)" + result.info, Constants.TIPS_TYPE_INFO) # 获取存档 PublicFunc.progressbar_update(self, 2, 7, Constants.LOADING_ARCHIVE, 20) result = self.data.get_archive() if not result.status: PublicFunc.text_output(self.ui, "(2/7)" + 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/7)" + result.info, Constants.TIPS_TYPE_INFO) # 数据预处理 PublicFunc.progressbar_update(self, 3, 7, Constants.PREPROCESSING_DATA, 30) result = self.data.preprocess() if not result.status: PublicFunc.text_output(self.ui, "(3/7)" + 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/7)" + result.info, Constants.TIPS_TYPE_INFO) # 重采样 PublicFunc.progressbar_update(self, 4, 7, Constants.RESAMPLING_DATA, 50) result = self.data.resample() for key, value in self.data.channel.items(): PublicFunc.text_output(self.ui, key + "重采样后的长度：" + str(len(value)), Constants.TIPS_TYPE_INFO) if not result.status: PublicFunc.text_output(self.ui, "(4/7)" + 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, "(4/7)" + result.info, Constants.TIPS_TYPE_INFO) # 绘图 PublicFunc.progressbar_update(self, 5, 7, Constants.DRAWING_DATA, 70) result = self.__plot__() if not result.status: PublicFunc.text_output(self.ui, "(5/7)" + 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, "(5/7)" + result.info, Constants.TIPS_TYPE_INFO) # 更新表格 PublicFunc.progressbar_update(self, 6, 7, Constants.UPDATING_TABLEWIDGET, 90) result = self.update_tableWidget() if not result.status: PublicFunc.text_output(self.ui, "(6/7)" + 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, "(6/7)" + result.info, Constants.TIPS_TYPE_INFO) # 更新信息 PublicFunc.progressbar_update(self, 7, 7, Constants.UPDATING_INFO, 95) result = self.update_UI_Args() if not result.status: PublicFunc.text_output(self.ui, "(7/7)" + 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, "(7/7)" + result.info, Constants.TIPS_TYPE_INFO) if (self.data.df_corrected["isLabeled"] == 1).all(): self.ui.checkBox_examineLabeled.setChecked(False) PublicFunc.msgbox_output(self, Constants.SA_LABEL_ALL_LABELED, Constants.MSGBOX_TYPE_INFO) self.__reset__() self.figToolbar.action_Reset_Signal_and_Time.setEnabled(True) for action in self.figToolbar._actions.values(): action.setEnabled(True) ButtonState["Current"]["pushButton_input"] = False ButtonState["Current"]["pushButton_input_setting"] = False ButtonState["Current"]["pushButton_quick_remark_input_waitingForTalk"] = True ButtonState["Current"]["pushButton_quick_remark_input_maybeDesaturation"] = True ButtonState["Current"]["pushButton_quick_remark_input_maybeWrongLabeled"] = True ButtonState["Current"]["pushButton_quick_remark_input_durationNoEnough"] = True ButtonState["Current"]["pushButton_quick_remark_input_littleChange"] = True ButtonState["Current"]["pushButton_quick_remark_input_noNormalRespBetweenArtifact"] = True ButtonState["Current"]["pushButton_quick_remark_input_lowSignalNoiseRatio"] = True ButtonState["Current"]["pushButton_quick_remark_input_changeOnMiddle"] = True ButtonState["Current"]["pushButton_prev"] = True ButtonState["Current"]["pushButton_next"] = True ButtonState["Current"]["pushButton_confirmLabel"] = True ButtonState["Current"]["pushButton_previous10s"] = False ButtonState["Current"]["pushButton_previous30s"] = False ButtonState["Current"]["pushButton_previous60s"] = False ButtonState["Current"]["pushButton_next10s"] = False ButtonState["Current"]["pushButton_next30s"] = False ButtonState["Current"]["pushButton_next60s"] = False PublicFunc.finish_operation(self, ButtonState) def __slot_btn_confirmLabel__(self): try: if (int(self.ui.spinBox_correctStart.value()) * Config["InputConfig"]["PlotFreq"] > len(self.data.channel['orgdata']) or int(self.ui.spinBox_correctStart.value()) < 0 or int(self.ui.spinBox_correctEnd.value()) < 0 or int(self.ui.spinBox_correctStart.value()) * Config["InputConfig"]["PlotFreq"] > len(self.data.channel['orgdata'])): PublicFunc.msgbox_output(self, Constants.SA_LABEL_LENGTH_TOO_LONG, Constants.MSGBOX_TYPE_ERROR) return if int(self.ui.spinBox_correctEnd.value()) - int(self.ui.spinBox_correctStart.value()) < 10: PublicFunc.msgbox_output(self, Constants.SA_LABEL_LENGTH_LESS_THEN_10S, Constants.MSGBOX_TYPE_WARNING) if self.ui.checkBox_examineBySecond.isChecked(): if int(self.ui.spinBox_correctStart.value()) < int(self.ui.spinBox_correctEnd.value()): if self.ui.radioButton_1_class.isChecked(): score = int(1) elif self.ui.radioButton_2_class.isChecked(): score = int(2) elif self.ui.radioButton_3_class.isChecked(): score = int(3) else: raise ValueError("score值不存在") remark = self.ui.lineEdit_remark.text() correct_Start = int(self.ui.spinBox_correctStart.value()) correct_End = int(self.ui.spinBox_correctEnd.value()) if self.ui.radioButton_OSA.isChecked(): correct_EventsType = "Obstructive apnea" elif self.ui.radioButton_CSA.isChecked(): correct_EventsType = "Central apnea" elif self.ui.radioButton_MSA.isChecked(): correct_EventsType = "Mixed apnea" elif self.ui.radioButton_HPY.isChecked(): correct_EventsType = "Hypopnea" else: raise ValueError("correct_EventsType值不存在") isLabeled = int(1) self.data.df_addNew = self.pd_add_new_row(self.data.df_addNew, score, remark, correct_Start, correct_End, correct_EventsType, isLabeled) result = self.data.save_2() info = f"保存新事件打标结果到csv。score:{str(score)}，correct_Start:{str(correct_Start)}，correct_End:{str(correct_End)}，correct_EventsType:{str(correct_EventsType)}" PublicFunc.text_output(self.ui, info, Constants.MSGBOX_TYPE_INFO) else: PublicFunc.msgbox_output(self, Constants.SA_LABEL_WRONG_ARGS, Constants.MSGBOX_TYPE_ERROR) else: if int(self.ui.spinBox_correctStart.value()) < int(self.ui.spinBox_correctEnd.value()): if self.ui.radioButton_1_class.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"] = int(1) elif self.ui.radioButton_2_class.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"] = int(2) elif self.ui.radioButton_3_class.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"] = int(3) self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "remark"] = self.ui.lineEdit_remark.text() self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"] = int( self.ui.spinBox_correctStart.value()) self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"] = int( self.ui.spinBox_correctEnd.value()) if self.ui.radioButton_OSA.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "correct_EventsType"] = "Obstructive apnea" elif self.ui.radioButton_CSA.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "correct_EventsType"] = "Central apnea" elif self.ui.radioButton_MSA.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "correct_EventsType"] = "Mixed apnea" elif self.ui.radioButton_HPY.isChecked(): self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "correct_EventsType"] = "Hypopnea" self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "isLabeled"] = int(1) # self.data.df_corrected.to_csv( # self.BCG_Label_Path / f"export{self.mainWindow.ui_SA_label_left.comboBox_sampID.currentText()}_all.csv", # mode='w', index=None, encoding="gbk") result = self.data.save() index = str(Config["PlotEventIndex"] + 1) score = str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "score"]) correct_Start = str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"]) correct_End = str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"]) correct_EventsType = str(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_EventsType"]) info = f"保存index{index}标注结果到csv。score:{score}，correct_Start:{correct_Start}，correct_End:{correct_End}，correct_EventsType:{correct_EventsType}" PublicFunc.text_output(self.ui, info, Constants.MSGBOX_TYPE_INFO) if (self.data.df_corrected.loc[Config["EventLabelIndexList"]]["isLabeled"] == 1).all(): self.ui.checkBox_examineLabeled.setChecked(False) PublicFunc.text_output(self.ui, Constants.SA_LABEL_ALL_LABELED, Constants.TIPS_TYPE_INFO) PublicFunc.msgbox_output(self, Constants.SA_LABEL_ALL_LABELED, Constants.MSGBOX_TYPE_INFO) else: PublicFunc.msgbox_output(self, Constants.SA_LABEL_WRONG_ARGS, Constants.MSGBOX_TYPE_ERROR) self.update_tableWidget() except ValueError as e: info = Constants.SAVE_FAILURE + "\n" + format_exc() PublicFunc.msgbox_output(self, info, Constants.MSGBOX_TYPE_ERROR) def __slot_btn_move__(self): if self.data is None: return sender = self.sender() if sender == self.ui.pushButton_prev: if Config["PlotEventIndex"] == 0: PublicFunc.msgbox_output(self, Constants.SA_LABEL_VIEWING_FIRST, Constants.MSGBOX_TYPE_INFO) return else: Config["PlotEventIndex"] = Config["PlotEventIndex"] - 1 if self.ui.checkBox_examineLabeled.isChecked(): while self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "isLabeled"] == 1 and Config["PlotEventIndex"] > 0: Config["PlotEventIndex"] = Config["PlotEventIndex"] - 1 if Config["PlotEventIndex"] == 0: PublicFunc.msgbox_output(self, Constants.SA_LABEL_VIEWING_FIRST, Constants.MSGBOX_TYPE_INFO) break self.__plot__() self.update_UI_Args() PublicFunc.text_output(self.ui, Constants.SA_LABEL_PREV_EVENT + str(Config["PlotEventIndex"] + 1), Constants.TIPS_TYPE_INFO) elif sender == self.ui.pushButton_next: if Config["PlotEventIndex"] == len(self.data.df_corrected) - 1: PublicFunc.msgbox_output(self, Constants.SA_LABEL_VIEWING_LAST, Constants.MSGBOX_TYPE_INFO) return else: Config["PlotEventIndex"] = Config["PlotEventIndex"] + 1 if self.ui.checkBox_examineLabeled.isChecked(): while self.data.df_corrected.at[Config["EventLabelIndexList"][ Config["PlotEventIndex"]], "isLabeled"] == 1 and Config["PlotEventIndex"] < len( self.data.df_corrected) - 1: Config["PlotEventIndex"] = Config["PlotEventIndex"] + 1 if Config["PlotEventIndex"] == len(self.data.df_corrected) - 1: PublicFunc.msgbox_output(self, Constants.SA_LABEL_VIEWING_LAST, Constants.MSGBOX_TYPE_INFO) break self.__plot__() self.update_UI_Args() PublicFunc.text_output(self.ui, Constants.SA_LABEL_NEXT_EVENT + str(Config["PlotEventIndex"] + 1), Constants.TIPS_TYPE_INFO) def __slot_btn_moveBySecond__(self): sender = self.sender() self.ui.lineEdit_remark.setText("") self.ui.radioButton_OSA.setChecked(True) if sender == self.ui.pushButton_previous10s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_PREV10S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] - 10 self.__plot__(Config["TimeMoveCount"]) elif sender == self.ui.pushButton_previous30s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_PREV30S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] - 30 self.__plot__(Config["TimeMoveCount"]) elif sender == self.ui.pushButton_previous60s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_PREV60S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] - 60 self.__plot__(Config["TimeMoveCount"]) elif sender == self.ui.pushButton_next10s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_NEXT10S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] + 10 self.__plot__(Config["TimeMoveCount"]) elif sender == self.ui.pushButton_next30s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_NEXT30S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] + 30 self.__plot__(Config["TimeMoveCount"]) elif sender == self.ui.pushButton_next60s: PublicFunc.text_output(self.ui, Constants.SA_LABEL_MOVE_NEXT60S, Constants.MSGBOX_TYPE_INFO) Config["TimeMoveCount"] = Config["TimeMoveCount"] + 60 self.__plot__(Config["TimeMoveCount"]) def __slot_btn_quick_remark__(self): sender = self.sender() if sender == self.ui.pushButton_quick_remark_input_waitingForTalk: self.ui.lineEdit_remark.setText("待讨论") elif sender == self.ui.pushButton_quick_remark_input_maybeDesaturation: self.ui.lineEdit_remark.setText("形似潮式呼吸") self.ui.radioButton_1_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_maybeWrongLabeled: self.ui.lineEdit_remark.setText("疑似医生误标") self.ui.radioButton_2_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_durationNoEnough: self.ui.lineEdit_remark.setText("时长不足") self.ui.radioButton_2_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_littleChange: self.ui.lineEdit_remark.setText("起伏变化不大") self.ui.radioButton_2_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_noNormalRespBetweenArtifact: self.ui.lineEdit_remark.setText("体动间无正常呼吸") self.ui.radioButton_2_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_lowSignalNoiseRatio: self.ui.lineEdit_remark.setText("信噪比低") self.ui.radioButton_2_class.setChecked(True) elif sender == self.ui.pushButton_quick_remark_input_changeOnMiddle: self.ui.lineEdit_remark.setText("中间起伏") self.ui.radioButton_2_class.setChecked(True) def __slot_tableWidget_label_on_cell_double_clicked__(self, row, column): Config["PlotEventIndex"] = int(row) PublicFunc.text_output(self.ui, "{}，index {}".format(Constants.SA_LABEL_JUMP, Config["PlotEventIndex"] + 1), Constants.TIPS_TYPE_INFO) self.__plot__() self.update_UI_Args() def __slot_lineEdit_filter_label__(self, filter_text): sender = self.sender() if sender == self.ui.lineEdit_filter_label: for row in range(self.ui.tableWidget_label.rowCount()): match = False for col in range(self.ui.tableWidget_label.columnCount()): item = self.ui.tableWidget_label.item(row, col) if filter_text.lower() in item.text().lower(): match = True break self.ui.tableWidget_label.setRowHidden(row, not match) elif sender == self.ui.lineEdit_filter_label_add: for row in range(self.ui.tableWidget_label_add.rowCount()): match = False for col in range(self.ui.tableWidget_label_add.columnCount()): item = self.ui.tableWidget_label_add.item(row, col) if filter_text.lower() in item.text().lower(): match = True break self.ui.tableWidget_label_add.setRowHidden(row, not match) else: raise ValueError("信号发生器错误") def __slot_checkBox_examineLabeled__(self): if self.ui.checkBox_examineLabeled.isChecked(): if (self.data.df_corrected.loc[Config["EventLabelIndexList"]]["isLabeled"] == 1).all(): self.ui.checkBox_examineLabeled.setChecked(False) PublicFunc.text_output(self.ui, Constants.SA_LABEL_ALL_LABELED, Constants.TIPS_TYPE_INFO) PublicFunc.msgbox_output(self, Constants.SA_LABEL_ALL_LABELED, Constants.MSGBOX_TYPE_INFO) def __slot_checkBox_examineBySecond__(self): if self.ui.checkBox_examineBySecond.isChecked(): ButtonState["Current"]["pushButton_previous10s"] = True ButtonState["Current"]["pushButton_previous30s"] = True ButtonState["Current"]["pushButton_previous60s"] = True ButtonState["Current"]["pushButton_next10s"] = True ButtonState["Current"]["pushButton_next30s"] = True ButtonState["Current"]["pushButton_next60s"] = True ButtonState["Current"]["pushButton_prev"]: False ButtonState["Current"]["pushButton_next"]: False self.ui.checkBox_examineLabeled.setEnabled(False) self.ui.radioButton_OSA.setChecked(True) self.ui.radioButton_2_class.setChecked(True) self.ui.radioButton_3_class.setEnabled(False) self.ui.lineEdit_remark.setText("") else: ButtonState["Current"]["pushButton_previous10s"] = False ButtonState["Current"]["pushButton_previous30s"] = False ButtonState["Current"]["pushButton_previous60s"] = False ButtonState["Current"]["pushButton_next10s"] = False ButtonState["Current"]["pushButton_next30s"] = False ButtonState["Current"]["pushButton_next60s"] = False self.ui.checkBox_examineLabeled.setEnabled(True) self.ui.radioButton_3_class.setEnabled(True) PublicFunc.finish_operation(self, ButtonState) def toggle_resetOriginalView(self): self.figToolbar.home() self.figToolbar.action_Reset_Signal_and_Time.setChecked(False) self.ui.spinBox_correctStart.setValue(Config["BCG_SP"]) self.ui.spinBox_correctEnd.setValue(Config["BCG_EP"]) def reset_axes(self): if self.ax0 is not None: self.ax0.clear() self.ax0.grid(True) self.ax0.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax0.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax1 is not None: self.ax1.clear() self.ax1.grid(True) self.ax1.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax1.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax2 is not None: self.ax2.clear() self.ax2.grid(True) self.ax2.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax2.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax3 is not None: self.ax3.clear() self.ax3.grid(True) self.ax3.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax3.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax4 is not None: self.ax4.clear() self.ax4.grid(True) self.ax4.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax4.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax5 is not None: self.ax5.clear() self.ax5.grid(True) self.ax5.xaxis.set_major_formatter(ConfigParams.FORMATTER) self.ax5.tick_params(axis='x', colors=Constants.PLOT_COLOR_WHITE) if self.ax6 is not None: self.ax6.clear() self.ax6.grid(True) self.ax6.xaxis.set_major_formatter(ConfigParams.FORMATTER) def on_xlim_change(self, event_ax): # 获取当前x轴范围 left, right = event_ax.get_xlim() self.ui.spinBox_correctStart.setValue(round(left)) self.ui.spinBox_correctEnd.setValue(round(right)) def draw_one_event(self): one_bcg_data = self.data.df_corrected.loc[Config["PlotEventIndex"]] # 获取BCG事件开始与结束时间 bcg_SP = one_bcg_data["Start"] bcg_EP = one_bcg_data["End"] Config["BCG_SP"] = bcg_SP Config["BCG_EP"] = bcg_EP bcg_duration = bcg_EP - bcg_SP # 进行向两边延展 bcg_SP = bcg_SP - Config["AddSecond"]["Front"] bcg_EP = bcg_EP + Config["AddSecond"]["Back"] if not (self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "isLabeled"] == -1): bcg_duration_new = self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"] - \ self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"] self.ui.spinBox_correctStart.setValue(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"]) self.ui.spinBox_correctEnd.setValue(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"]) self.ui.label_BCG_event.setText("Index:{}/{}".format((Config["PlotEventIndex"] + 1), len(self.data.df_corrected))) self.ui.label_BCG_event_2.setText("Epoch:{} Duration:{}s New Duration:{}s".format(one_bcg_data["Epoch"], bcg_duration, bcg_duration_new)) else: bcg_duration_new = -1 self.ui.spinBox_correctStart.setValue(Config["BCG_SP"]) self.ui.spinBox_correctEnd.setValue(Config["BCG_EP"]) self.ui.label_BCG_event.setText("Index:{}/{}".format((Config["PlotEventIndex"] + 1), len(self.data.df_corrected))) self.ui.label_BCG_event_2.setText("Epoch:{} Duration:{}s".format(one_bcg_data["Epoch"], bcg_duration)) self.plt_channel(plt_=self.ax0, SP=bcg_SP, EP=bcg_EP, channel="Flow T", bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax1, SP=bcg_SP, EP=bcg_EP, channel="Flow P", bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax2, SP=bcg_SP, EP=bcg_EP, channel="Effort Tho", bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax3, SP=bcg_SP, EP=bcg_EP, channel="Effort Abd", bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax4, SP=bcg_SP, EP=bcg_EP, channel="SpO2", event_code=[5], bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax5, SP=bcg_SP, EP=bcg_EP, channel="orgdata", event_code=[6, 7, 8, 9, 10, 1, 2, 3, 4], event_show_under=False, bcg_duration_new=bcg_duration_new, show_mode="one") self.plt_channel(plt_=self.ax6, SP=bcg_SP, EP=bcg_EP, channel="0.7lowpass_resp", event_code=[6, 7, 8, 9, 10, 1, 2, 3, 4], event_show_under=False, bcg_duration_new=bcg_duration_new, show_mode="one") self.ax0.callbacks.connect('xlim_changed', lambda ax: self.on_xlim_change(ax)) def draw_new_event(self, time_move_count: int): one_bcg_data = self.data.df_corrected.loc[Config["PlotEventIndex"]] # 获取BCG事件开始与结束时间 bcg_SP = one_bcg_data["Start"] bcg_EP = one_bcg_data["End"] Config["BCG_SP"] = bcg_SP Config["BCG_EP"] = bcg_EP bcg_duration = bcg_EP - bcg_SP # 进行向两边延展 bcg_SP = bcg_SP - Config["AddSecond"]["Front"] + time_move_count bcg_EP = bcg_EP + Config["AddSecond"]["Back"] + time_move_count self.plt_channel(plt_=self.ax0, SP=bcg_SP, EP=bcg_EP, channel="Flow T", show_mode="new") self.plt_channel(plt_=self.ax1, SP=bcg_SP, EP=bcg_EP, channel="Flow P", show_mode="new") self.plt_channel(plt_=self.ax2, SP=bcg_SP, EP=bcg_EP, channel="Effort Tho", show_mode="new") self.plt_channel(plt_=self.ax3, SP=bcg_SP, EP=bcg_EP, channel="Effort Abd", show_mode="new") self.plt_channel(plt_=self.ax4, SP=bcg_SP, EP=bcg_EP, channel="SpO2", event_code=[5], show_mode="new") self.plt_channel(plt_=self.ax5, SP=bcg_SP, EP=bcg_EP, channel="orgdata", event_code=[6, 7, 8, 9, 10, 1, 2, 3, 4], event_show_under=False, show_mode="new") self.plt_channel(plt_=self.ax6, SP=bcg_SP, EP=bcg_EP, channel="0.7lowpass_resp", event_code=[6, 7, 8, 9, 10, 1, 2, 3, 4], event_show_under=False, show_mode="new") self.ui.label_BCG_event.setText( "Index:{}/{}".format((Config["PlotEventIndex"] + 1), len(self.data.df_corrected))) self.ui.label_BCG_event_2.setText("Epoch:{} Duration:{}s".format(one_bcg_data["Epoch"], bcg_duration)) self.ax0.callbacks.connect('xlim_changed', lambda ax: self.on_xlim_change(ax)) def plt_channel(self, plt_, SP, EP, channel, event_code=[1, 2, 3, 4], event_show_under=False, bcg_duration_new=-1, show_mode=None): SP = 0 if SP < 0 else SP plt_.plot(linspace(SP, EP, (EP - SP) * Config["InputConfig"]["PlotFreq"]), self.data.channel[channel][SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]], label=channel, color=self.color_cycle[0]) for j in event_code: if channel == "SpO2": mask = self.data.event_label[SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]] == j elif channel == "orgdata" or channel == "0.7lowpass_resp": if j <= 5: mask = self.data.event_label[SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]] == j else: mask = self.data.artifact_label[SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]] == j else: mask = self.data.event_label[SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]] == j if event_show_under: min_point = self.data.channel[channel][SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]].min() len_segment = EP * Config["InputConfig"]["PlotFreq"] - SP * Config["InputConfig"]["PlotFreq"] y = (min_point.repeat(len_segment) * mask).astype('float') place(y, y == 0, nan) else: y = (self.data.channel[channel][SP * Config["InputConfig"]["PlotFreq"]:EP * Config["InputConfig"]["PlotFreq"]] * mask).astype('float') place(y, y == 0, nan) plt_.plot(linspace(SP, EP, (EP - SP) * Config["InputConfig"]["PlotFreq"]), y, color=self.color_cycle[j], linestyle="-") plt_.legend(fontsize=8, loc=1) ax_ = [self.ax0, self.ax1, self.ax2, self.ax3, self.ax4, self.ax5, self.ax6] # 绘制半透明矩形框 if show_mode == "one": if not (bcg_duration_new == -1): if self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_EventsType"] == "Obstructive apnea": for ax in ax_: ax.axvspan( self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"], color='red', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_EventsType"] == "Central apnea": for ax in ax_: ax.axvspan( self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"], color='blue', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_EventsType"] == "Mixed apnea": for ax in ax_: ax.axvspan( self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"], color='gray', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_EventsType"] == "Hypopnea": for ax in ax_: ax.axvspan( self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "correct_End"], color='pink', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) else: if self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Event type"] == "Obstructive apnea": for ax in ax_: ax.axvspan(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "End"], color='red', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Event type"] == "Central apnea": for ax in ax_: ax.axvspan(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "End"], color='blue', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Event type"] == "Mixed apnea": for ax in ax_: ax.axvspan(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "End"], color='gray', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) elif self.data.df_corrected.at[ Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Event type"] == "Hypopnea": for ax in ax_: ax.axvspan(self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "Start"], self.data.df_corrected.at[Config["EventLabelIndexList"][Config["PlotEventIndex"]], "End"], color='pink', alpha=ConfigParams.SA_LABEL_TRANSPARENCY) def pd_add_new_row(self, df, score: int, remark: str, correct_Start: int, correct_End: int, correct_EventsType: str, isLabeled: int): new_row = Series(index=df.columns, data=nan) new_row["score"] = score new_row["remark"] = remark new_row["correct_Start"] = correct_Start new_row["correct_End"] = correct_End new_row["correct_EventsType"] = correct_EventsType new_row["isLabeled"] = isLabeled return concat([df, DataFrame([new_row])], ignore_index=True) class Data: def __init__(self): self.OrgBCG = None self.Tho = None self.Abd = None self.FlowT = None self.FlowP = None self.SpO2 = None self.lowPass20Hz = None self.lowPassResp = None self.OrgBCG_Resampled = None self.Tho_Resampled = None self.Abd_Resampled = None self.FlowT_Resampled = None self.FlowP_Resampled = None self.SpO2_Resampled = None self.lowPass20Hz_Resampled = None self.lowPassResp_Resampled = None self.channel = { "orgdata": array([]), "0.7lowpass_resp": array([]), "Effort Tho": array([]), "Effort Abd": array([]), "Flow T": array([]), "Flow P": array([]), "SpO2": array([]) } self.event_label = None self.artifact_label = None self.Artifact = None self.df_corrected = None self.df_addNew = None def open_file(self): if Path(Config["Path"]["Input_OrgBCG"]).is_file(): Config["Path"]["Input_OrgBCG"] = str(Path(Config["Path"]["Input_orgBcg"]).parent) if Path(Config["Path"]["Input_Tho"]).is_file(): Config["Path"]["Input_Tho"] = str(Path(Config["Path"]["Input_Tho"]).parent) if Path(Config["Path"]["Input_Abd"]).is_file(): Config["Path"]["Input_Abd"] = str(Path(Config["Path"]["Input_Abd"]).parent) if Path(Config["Path"]["Input_FlowT"]).is_file(): Config["Path"]["Input_FlowT"] = str(Path(Config["Path"]["Input_FlowT"]).parent) if Path(Config["Path"]["Input_FlowP"]).is_file(): Config["Path"]["Input_FlowP"] = str(Path(Config["Path"]["Input_FlowP"]).parent) if Path(Config["Path"]["Input_SpO2"]).is_file(): Config["Path"]["Input_SpO2"] = str(Path(Config["Path"]["Input_SpO2"]).parent) if Path(Config["Path"]["Input_Artifact"]).is_file(): Config["Path"]["Input_Artifact"] = str(Path(Config["Path"]["Input_Artifact"]).parent) if Path(Config["Path"]["Input_Label"]).is_file(): Config["Path"]["Input_Label"] = str(Path(Config["Path"]["Input_Label"]).parent) if ((not Path(Config["Path"]["Input_OrgBCG"]).exists()) or (not Path(Config["Path"]["Input_Tho"]).exists()) or (not Path(Config["Path"]["Input_Abd"]).exists()) or (not Path(Config["Path"]["Input_FlowT"]).exists()) or (not Path(Config["Path"]["Input_FlowP"]).exists()) or (not Path(Config["Path"]["Input_SpO2"]).exists()) or (not Path(Config["Path"]["Input_Artifact"]).exists())): return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Path_Not_Exist"]) result = PublicFunc.examine_file(Config["Path"]["Input_OrgBCG"], ConfigParams.ORGBCG_SYNC) if result.status: Config["Path"]["Input_OrgBCG"] = result.data["path"] Config["InputConfig"]["OrgBCGFreq"] = result.data["freq"] else: return result result = PublicFunc.examine_file(Config["Path"]["Input_Tho"], ConfigParams.THO_SYNC) if result.status: Config["Path"]["Input_Tho"] = result.data["path"] Config["InputConfig"]["ThoFreq"] = result.data["freq"] else: return result result = PublicFunc.examine_file(Config["Path"]["Input_Abd"], ConfigParams.ABD_SYNC) if result.status: Config["Path"]["Input_Abd"] = result.data["path"] Config["InputConfig"]["AbdFreq"] = result.data["freq"] else: return result result = PublicFunc.examine_file(Config["Path"]["Input_FlowT"], ConfigParams.FLOWT_SYNC) if result.status: Config["Path"]["Input_FlowT"] = result.data["path"] Config["InputConfig"]["FlowTFreq"] = result.data["freq"] else: return result result = PublicFunc.examine_file(Config["Path"]["Input_FlowP"], ConfigParams.FLOWP_SYNC) if result.status: Config["Path"]["Input_FlowP"] = result.data["path"] Config["InputConfig"]["FlowPFreq"] = result.data["freq"] else: return result result = PublicFunc.examine_file(Config["Path"]["Input_SpO2"], ConfigParams.SPO2_SYNC) if result.status: Config["Path"]["Input_SpO2"] = result.data["path"] Config["InputConfig"]["SpO2Freq"] = result.data["freq"] else: return result Config["Path"]["Input_Artifact"] = str( Path(Config["Path"]["Input_Artifact"]) / Path(ConfigParams.ARTIFACT_A + ConfigParams.ENDSWITH_TXT)) Config["Path"]["Input_Label"] = str( Path(Config["Path"]["Input_Label"]) / Path(ConfigParams.SA_LABEL_SYNC + ConfigParams.ENDSWITH_CSV)) Config["Path"]["Save"] = str( Path(Config["Path"]["Save"]) / Path(ConfigParams.SA_LABEL_CORRECTED + ConfigParams.ENDSWITH_CSV)) Config["Path"]["Save_2"] = str( Path(Config["Path"]["Save_2"]) / Path(ConfigParams.SA_LABEL_ADD + ConfigParams.ENDSWITH_CSV)) try: self.OrgBCG = read_csv(Config["Path"]["Input_OrgBCG"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.Tho = read_csv(Config["Path"]["Input_Tho"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.Abd = read_csv(Config["Path"]["Input_Abd"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.FlowT = read_csv(Config["Path"]["Input_FlowT"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.FlowP = read_csv(Config["Path"]["Input_FlowP"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.SpO2 = read_csv(Config["Path"]["Input_SpO2"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.Artifact = read_csv(Config["Path"]["Input_Artifact"], encoding=ConfigParams.UTF8_ENCODING, header=None).to_numpy().reshape(-1) self.event_label = zeros(len(self.OrgBCG) + (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"])) except Exception as e: return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Open_Data_Exception"] + "\n" + format_exc()) try: # 检查体动标签正确性，长度 if len(self.Artifact) % 4 != 0: return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Artifact_Format_Not_Correct"]) for i in range(0, len(self.Artifact), 4): unit_data = self.Artifact[i:i + 4] if len(unit_data) < 4: break self.Artifact = self.Artifact.reshape(-1, 4) except Exception as e: return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Get_Artifact_Format_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.INPUT_FINISHED) def get_archive(self): if (not Path(Config["Path"]["Save"]).exists()) or (not Path(Config["Path"]["Save_2"]).exists()): self.df_corrected = read_csv(Config["Path"]["Input_Label"], encoding="gbk") self.df_corrected["Start"] = self.df_corrected["Start"].astype(int) self.df_corrected["End"] = self.df_corrected["End"].astype(int) self.df_corrected["score"] = "-1" self.df_corrected["remark"] = "" self.df_corrected["correct_Start"] = "-1" self.df_corrected["correct_End"] = "-1" self.df_corrected["correct_EventsType"] = "" self.df_corrected["isLabeled"] = "-1" self.df_corrected["score"] = self.df_corrected["score"].astype(int) self.df_corrected["remark"] = self.df_corrected["remark"].astype(str) self.df_corrected["correct_Start"] = self.df_corrected["correct_Start"].astype(int) self.df_corrected["correct_End"] = self.df_corrected["correct_End"].astype(int) self.df_corrected["correct_EventsType"] = self.df_corrected["correct_EventsType"].astype(str) self.df_corrected["isLabeled"] = self.df_corrected["isLabeled"].astype(int) self.df_corrected = self.df_corrected.sort_values(by='Epoch') for one_data in self.df_corrected.index: one_data = self.df_corrected.loc[one_data] SP = one_data["Start"] * Config["InputConfig"]["PlotFreq"] EP = one_data["End"] * Config["InputConfig"]["PlotFreq"] if one_data["Event type"] == "Hypopnea": self.event_label[SP:EP] = 1 elif one_data["Event type"] == "Central apnea": self.event_label[SP:EP] = 2 elif one_data["Event type"] == "Obstructive apnea": self.event_label[SP:EP] = 3 elif one_data["Event type"] == "Mixed apnea": self.event_label[SP:EP] = 4 self.df_addNew = DataFrame(columns=self.df_corrected.columns) self.df_addNew["score"] = self.df_addNew["score"].astype(int) self.df_addNew["remark"] = self.df_addNew["remark"].astype(str) self.df_addNew["correct_Start"] = self.df_addNew["correct_Start"].astype(int) self.df_addNew["correct_End"] = self.df_addNew["correct_End"].astype(int) self.df_addNew["correct_EventsType"] = self.df_addNew["correct_EventsType"].astype( str) self.df_addNew["isLabeled"] = self.df_addNew["isLabeled"].astype(int) self.df_addNew.to_csv(Config["Path"]["Save_2"], index=False, encoding="gbk") Config["EventLabelIndexList"] = self.df_corrected.index.tolist() return Result().success(info=Constants.ARCHIVE_NOT_EXIST) else: try: self.df_corrected = read_csv(Config["Path"]["Save"], encoding="gbk") self.df_addNew = read_csv(Config["Path"]["Save_2"], encoding="gbk") csv_headers = self.df_corrected.columns.tolist() if not (("score" in csv_headers) or ("remark" in csv_headers) or ("correct_Start" in csv_headers) or ( "correct_End") in csv_headers or ("isLabeled" in csv_headers)): return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON[ "Label_Format_Exception"]) csv_headers = self.df_addNew.columns.tolist() if not (("score" in csv_headers) or ("remark" in csv_headers) or ("correct_Start" in csv_headers) or ( "correct_End") in csv_headers or ("isLabeled" in csv_headers)): return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON[ "Label_Format_Exception"]) Config["EventLabelIndexList"] = self.df_corrected.index.tolist() for one_data in self.df_corrected.index: one_data = self.df_corrected.loc[one_data] SP = one_data["Start"] * Config["InputConfig"]["PlotFreq"] EP = one_data["End"] * Config["InputConfig"]["PlotFreq"] if one_data["Event type"] == "Hypopnea": self.event_label[SP:EP] = 1 elif one_data["Event type"] == "Central apnea": self.event_label[SP:EP] = 2 elif one_data["Event type"] == "Obstructive apnea": self.event_label[SP:EP] = 3 elif one_data["Event type"] == "Mixed apnea": self.event_label[SP:EP] = 4 except Exception as e: return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON[ "Label_Format_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.ARCHIVE_EXIST) def preprocess(self): if self.OrgBCG is None: return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"]) try: self.lowPass20Hz = Butterworth_for_ECG_PreProcess(self.OrgBCG, Config["InputConfig"]["OrgBCGFreq"], 'lowpass', low_cut=20, order=3) self.lowPassResp = Butterworth_for_ECG_PreProcess(self.OrgBCG, Config["InputConfig"]["OrgBCGFreq"], 'lowpass', low_cut=0.7, order=3) self.artifact_label = zeros(len(self.event_label) + (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"])) for i, artifact_type, SP, EP in self.Artifact: SP = int(SP) // (Config["InputConfig"]["OrgBCGFreq"] // Config["InputConfig"]["PlotFreq"]) EP = int(EP) // (Config["InputConfig"]["OrgBCGFreq"] // Config["InputConfig"]["PlotFreq"]) artifact_type = int(artifact_type) + 5 SP = 0 if SP < 0 else SP if EP < 0: continue if SP > len(self.event_label): continue self.artifact_label[SP:EP] = artifact_type except Exception as e: return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Preprocess_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.PREPROCESS_FINISHED) def resample(self): if self.Tho is None or self.Abd is None or self.FlowT is None or self.FlowP is None or self.SpO2 is None or self.lowPass20Hz is None or self.lowPassResp is None: return Result().failure(info=Constants.RESAMPLE_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"]) try: if Config["InputConfig"]["OrgBCGFreq"] < Config["InputConfig"]["PlotFreq"]: self.lowPass20Hz_Resampled = self.lowPass20Hz.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["OrgBCGFreq"]) self.lowPassResp_Resampled = self.lowPassResp.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["OrgBCGFreq"]) elif Config["InputConfig"]["OrgBCGFreq"] > Config["InputConfig"]["PlotFreq"]: self.lowPass20Hz_Resampled = self.lowPass20Hz[::int( Config["InputConfig"]["OrgBCGFreq"] / Config["InputConfig"]["PlotFreq"])] self.lowPassResp_Resampled = self.lowPassResp[::int( Config["InputConfig"]["OrgBCGFreq"] / Config["InputConfig"]["PlotFreq"])] else: self.lowPass20Hz_Resampled = self.lowPass20Hz self.lowPassResp_Resampled = self.lowPassResp self.lowPass20Hz_Resampled = append(self.lowPass20Hz_Resampled, self.lowPass20Hz_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) self.lowPassResp_Resampled = append(self.lowPassResp_Resampled, self.lowPassResp_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) if Config["InputConfig"]["ThoFreq"] < Config["InputConfig"]["PlotFreq"]: self.Tho_Resampled = self.Tho.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["ThoFreq"]) elif Config["InputConfig"]["ThoFreq"] > Config["InputConfig"]["PlotFreq"]: self.Tho_Resampled = self.Tho[::int( Config["InputConfig"]["ThoFreq"] / Config["InputConfig"]["PlotFreq"])] else: self.Tho_Resampled = self.Tho self.Tho_Resampled = append(self.Tho_Resampled, self.Tho_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) if Config["InputConfig"]["AbdFreq"] < Config["InputConfig"]["PlotFreq"]: self.Abd_Resampled = self.Abd.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["AbdFreq"]) elif Config["InputConfig"]["AbdFreq"] > Config["InputConfig"]["PlotFreq"]: self.Abd_Resampled = self.Abd[::int( Config["InputConfig"]["AbdFreq"] / Config["InputConfig"]["PlotFreq"])] else: self.Abd_Resampled = self.Abd self.Abd_Resampled = append(self.Abd_Resampled, self.Abd_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) if Config["InputConfig"]["FlowTFreq"] < Config["InputConfig"]["PlotFreq"]: self.FlowT_Resampled = self.FlowT.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["FlowTFreq"]) elif Config["InputConfig"]["FlowTFreq"] > Config["InputConfig"]["PlotFreq"]: self.FlowT_Resampled = self.FlowT[::int( Config["InputConfig"]["FlowTFreq"] / Config["InputConfig"]["PlotFreq"])] else: self.FlowT_Resampled = self.FlowT self.FlowT_Resampled = append(self.FlowT_Resampled, self.FlowT_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) if Config["InputConfig"]["FlowPFreq"] < Config["InputConfig"]["PlotFreq"]: self.FlowP_Resampled = self.FlowP.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["FlowPFreq"]) elif Config["InputConfig"]["FlowPFreq"] > Config["InputConfig"]["PlotFreq"]: self.FlowP_Resampled = self.FlowP[::int( Config["InputConfig"]["FlowPFreq"] / Config["InputConfig"]["PlotFreq"])] else: self.FlowP_Resampled = self.FlowP self.FlowP_Resampled = append(self.FlowP_Resampled, self.FlowP_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) if Config["InputConfig"]["SpO2Freq"] < Config["InputConfig"]["PlotFreq"]: self.SpO2_Resampled = self.SpO2.repeat( Config["InputConfig"]["PlotFreq"] / Config["InputConfig"]["SpO2Freq"]) elif Config["InputConfig"]["SpO2Freq"] > Config["InputConfig"]["PlotFreq"]: self.SpO2_Resampled = self.SpO2[::int( Config["InputConfig"]["SpO2Freq"] / Config["InputConfig"]["PlotFreq"])] else: self.SpO2_Resampled = self.SpO2 self.SpO2_Resampled = append(self.SpO2_Resampled, self.SpO2_Resampled.mean().astype(int).repeat( (int(Config["AddSecond"]["Front"] + int(Config["AddSecond"]["Back"])) * Config["InputConfig"]["PlotFreq"]))) self.channel["orgdata"] = self.lowPass20Hz_Resampled self.channel["0.7lowpass_resp"] = self.lowPassResp_Resampled self.channel["Effort Tho"] = self.Tho_Resampled self.channel["Effort Abd"] = self.Abd_Resampled self.channel["Flow T"] = self.FlowT_Resampled self.channel["Flow P"] = self.FlowP_Resampled self.channel["SpO2"] = self.SpO2_Resampled except Exception as e: return Result().failure(info=Constants.RESAMPLE_FAILURE + Constants.FAILURE_REASON["Resample_Exception"] + "\n" + format_exc()) return Result().success(info=Constants.RESAMPLE_FINISHED) def save(self): if (not Path(Config["Path"]["Save"]).parent.exists()) or (not Path(Config["Path"]["Save"]).parent.is_dir()): Path(Config["Path"]["Save"]).parent.mkdir(parents=True, exist_ok=True) if self.df_corrected is None: return Result().failure(info=Constants.SAVE_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"]) try: self.df_corrected.to_csv(Config["Path"]["Save"], mode='w', index=None, encoding="gbk") except Exception as e: return Result().failure(info=Constants.SAVE_FAILURE + Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc()) return Result().success(Constants.SAVE_FINISHED) def save_2(self): if (not Path(Config["Path"]["Save_2"]).parent.exists()) or (not Path(Config["Path"]["Save_2"]).parent.is_dir()): Path(Config["Path"]["Save_2"]).parent.mkdir(parents=True, exist_ok=True) if self.df_addNew is None: return Result().failure(info=Constants.SAVE_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"]) try: self.df_addNew.to_csv(Config["Path"]["Save_2"], mode='w', index=None, encoding="gbk") except Exception as e: return Result().failure(info=Constants.SAVE_FAILURE + Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc()) return Result().success(Constants.SAVE_FINISHED) class CustomNavigationToolbar(NavigationToolbar2QT): def __init__(self, canvas, parent): super().__init__(canvas, parent) self.action_Reset_Signal_and_Time = QAction('复原视图和时间', self) self.action_Reset_Signal_and_Time.setFont(QFont(ConfigParams.FONT, 14)) self.action_Reset_Signal_and_Time.setCheckable(True) self.insertAction(self._actions['home'], self.action_Reset_Signal_and_Time)