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

优化文件路径读取逻辑，增强异常处理，调整信号处理流程

Browse Source
This commit is contained in:
marques
2025-12-17 01:38:22 +08:00
parent c6161e7dd5
commit ea024eafbc
1 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
func/Module_SA_label_v2.py
View File

@ -1,3 +1,4 @@
import traceback
from gc import collect from gc import collect
from pathlib import Path from pathlib import Path
from traceback import format_exc from traceback import format_exc
@ -149,7 +150,7 @@ class SettingWindow(QMainWindow):
def __write_config__(self): def __write_config__(self):
# 从界面写入配置 # 从界面写入配置
self.params["Path"]["Input_Stage"] = self.ui.plainTextEdit_file_path_input_signal_Stage().toPlainText() self.params["Path"]["Input_Stage"] = self.ui.plainTextEdit_file_path_input_signal_Stage.toPlainText()
self.params["Path"]["Input_OrgBCG"] = self.ui.plainTextEdit_file_path_input_signal_OrgBCG.toPlainText() self.params["Path"]["Input_OrgBCG"] = self.ui.plainTextEdit_file_path_input_signal_OrgBCG.toPlainText()
self.params["Path"]["Input_Tho"] = self.ui.plainTextEdit_file_path_input_signal_Tho.toPlainText() self.params["Path"]["Input_Tho"] = self.ui.plainTextEdit_file_path_input_signal_Tho.toPlainText()
self.params["Path"]["Input_Abd"] = self.ui.plainTextEdit_file_path_input_signal_Abd.toPlainText() self.params["Path"]["Input_Abd"] = self.ui.plainTextEdit_file_path_input_signal_Abd.toPlainText()
@ -195,7 +196,8 @@ class SettingWindow(QMainWindow):
# self.ui.plainTextEdit_file_path_save_2.setPlainText(str(self.params["Path"]["Save_2"])) # self.ui.plainTextEdit_file_path_save_2.setPlainText(str(self.params["Path"]["Save_2"]))
def __auto_find_file__(self): def __auto_find_file__(self):
check_signal_type_list = ["Input_OrgBCG", "Input_Tho", "Input_Abd", "Input_FlowT", "Input_FlowP", "Input_SpO2", "Input_Stage"] check_signal_type_list = ["Input_OrgBCG", "Input_Tho", "Input_Abd", "Input_FlowT", "Input_FlowP", "Input_SpO2",
"Input_Stage"]
def find_file(file_path: Path, _type, endswith): def find_file(file_path: Path, _type, endswith):
if file_path.is_file(): if file_path.is_file():
@ -300,12 +302,12 @@ class Data:
for file_key in check_file_list: for file_key in check_file_list:
if (not self.config["Path"][file_key].is_file()) or (not self.config["Path"][file_key].exists()): if (not self.config["Path"][file_key].is_file()) or (not self.config["Path"][file_key].exists()):
return Result().failure(info=Constants.INPUT_FAILURE + "\n" + return Result().failure(info=Constants.INPUT_FAILURE + "\n" +
str(self.config["Path"][file_key]) + str(self.config["Path"][file_key]) + file_key +
Constants.FAILURE_REASON["Path_Not_Exist"]) Constants.FAILURE_REASON["Path_Not_Exist"])
try: try:
self.Stage = read_csv(self.config["Path"]["Input_Stage"], encoding=Params.UTF8_ENCODING, self.Stage = read_csv(self.config["Path"]["Input_Stage"], encoding=Params.UTF8_ENCODING,
header=None).to_numpy().reshape(-1) header=None).to_numpy().reshape(-1)
self.OrgBCG = read_csv(self.config["Path"]["Input_OrgBCG"], encoding=Params.UTF8_ENCODING, self.OrgBCG = read_csv(self.config["Path"]["Input_OrgBCG"], encoding=Params.UTF8_ENCODING,
header=None).to_numpy().reshape(-1) header=None).to_numpy().reshape(-1)
self.Tho = read_csv(self.config["Path"]["Input_Tho"], encoding=Params.UTF8_ENCODING, self.Tho = read_csv(self.config["Path"]["Input_Tho"], encoding=Params.UTF8_ENCODING,
@ -319,7 +321,6 @@ class Data:
self.SpO2 = read_csv(self.config["Path"]["Input_SpO2"], encoding=Params.UTF8_ENCODING, self.SpO2 = read_csv(self.config["Path"]["Input_SpO2"], encoding=Params.UTF8_ENCODING,
header=None).to_numpy().reshape(-1) header=None).to_numpy().reshape(-1)
if self.config["Path"]["Input_Artifact_A"].exists() and self.config["Path"]["Input_Artifact_A"].is_file(): if self.config["Path"]["Input_Artifact_A"].exists() and self.config["Path"]["Input_Artifact_A"].is_file():
self.Artifact = read_csv(self.config["Path"]["Input_Artifact_A"], self.Artifact = read_csv(self.config["Path"]["Input_Artifact_A"],
encoding=Params.UTF8_ENCODING, encoding=Params.UTF8_ENCODING,
@ -329,6 +330,7 @@ class Data:
except Exception as e: except Exception as e:
traceback.print_exc()
return Result().failure(info=Constants.INPUT_FAILURE + return Result().failure(info=Constants.INPUT_FAILURE +
Constants.FAILURE_REASON["Open_Data_Exception"] + "\n" + format_exc()) Constants.FAILURE_REASON["Open_Data_Exception"] + "\n" + format_exc())
@ -345,7 +347,6 @@ class Data:
"SignalSecond": int(len(self.OrgBCG) // self.config["Config"]["InputConfig"]["OrgBCGFreq"]) "SignalSecond": int(len(self.OrgBCG) // self.config["Config"]["InputConfig"]["OrgBCGFreq"])
}) })
# 批量将睡眠分期按照映射转换为数字 # 批量将睡眠分期按照映射转换为数字
for stage_str, stage_val in self.stage_to_value.items(): for stage_str, stage_val in self.stage_to_value.items():
place(self.Stage, self.Stage == stage_str, stage_val) place(self.Stage, self.Stage == stage_str, stage_val)
@ -360,7 +361,6 @@ class Data:
self.SpO2 = self.SpO2[:self.config["SignalSecond"] * self.config["Config"]["InputConfig"]["SpO2Freq"]] self.SpO2 = self.SpO2[:self.config["SignalSecond"] * self.config["Config"]["InputConfig"]["SpO2Freq"]]
self.Stage = self.Stage[:self.config["SignalSecond"]] self.Stage = self.Stage[:self.config["SignalSecond"]]
plot_freq = self.config["Config"]["InputConfig"]["PlotFreq"] plot_freq = self.config["Config"]["InputConfig"]["PlotFreq"]
self.event_label_origin = zeros(self.config["SignalSecond"] * plot_freq) self.event_label_origin = zeros(self.config["SignalSecond"] * plot_freq)
self.event_label_revised = zeros(self.config["SignalSecond"] * plot_freq) self.event_label_revised = zeros(self.config["SignalSecond"] * plot_freq)
@ -448,7 +448,10 @@ class Data:
def preprocess(self): def preprocess(self):
if self.OrgBCG is None: if self.OrgBCG is None:
return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"]) return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"])
if self.Tho is None:
return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"])
if self.Abd is None:
return Result().failure(info=Constants.PREPROCESS_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"])
try: try:
plot_freq = self.config["Config"]["InputConfig"]["PlotFreq"] plot_freq = self.config["Config"]["InputConfig"]["PlotFreq"]
orgbcg_freq = self.config["Config"]["InputConfig"]["OrgBCGFreq"] orgbcg_freq = self.config["Config"]["InputConfig"]["OrgBCGFreq"]
@ -459,6 +462,7 @@ class Data:
self.lowPassResp = Butterworth_for_ECG_PreProcess(self.OrgBCG, self.lowPassResp = Butterworth_for_ECG_PreProcess(self.OrgBCG,
orgbcg_freq, orgbcg_freq,
'lowpass', low_cut=0.7, order=3) 'lowpass', low_cut=0.7, order=3)
self.artifact_label = zeros(len(self.event_label_origin)) self.artifact_label = zeros(len(self.event_label_origin))
for i, artifact_type, start, end in self.Artifact: for i, artifact_type, start, end in self.Artifact:
@ -863,8 +867,10 @@ class MainWindow_SA_label(QMainWindow):
self.ui.pushButton_confirmLabel.clicked.connect(self.__slot_btn_confirmLabel__) self.ui.pushButton_confirmLabel.clicked.connect(self.__slot_btn_confirmLabel__)
self.ui.pushButton_reset_event.clicked.connect(self.__reset_event__) self.ui.pushButton_reset_event.clicked.connect(self.__reset_event__)
self.ui.pushButton_next_half.setProperty("offset", int(self.ui.comboBox_window_signal_length.lineEdit().text()) // 2) self.ui.pushButton_next_half.setProperty("offset",
self.ui.pushButton_previous_half.setProperty("offset", -int(self.ui.comboBox_window_signal_length.lineEdit().text()) // 2) int(self.ui.comboBox_window_signal_length.lineEdit().text()) // 2)
self.ui.pushButton_previous_half.setProperty("offset",
-int(self.ui.comboBox_window_signal_length.lineEdit().text()) // 2)
# 输入防抖 # 输入防抖
self.debounce_timer1 = QTimer() self.debounce_timer1 = QTimer()
@ -965,8 +971,6 @@ class MainWindow_SA_label(QMainWindow):
not self.channel_to_best_fit_checkbox[channel].isChecked()): not self.channel_to_best_fit_checkbox[channel].isChecked()):
continue continue
signal_max = self.data.channel[channel][start_point: end_point].max() signal_max = self.data.channel[channel][start_point: end_point].max()
signal_min = self.data.channel[channel][start_point: end_point].min() signal_min = self.data.channel[channel][start_point: end_point].min()
if channel == "SpO2": if channel == "SpO2":