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

修改了文件命名规则

Browse Source 
优化<ECG的R峰算法定位>的代码
This commit is contained in:
Yorusora
2025-05-20 17:16:26 +08:00
parent ec16546f47
commit d01dbcb5f5
4 changed files with 72 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
func/Module_detect_Rpeak.py
View File

@ -71,12 +71,9 @@ class SettingWindow(QMainWindow):
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))),
Path(str(self.sampID)))),
"Save": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.DETECT_RPEAK_SAVE_FILENAME +
ConfigParams.ENDSWITH_TXT)))
Path(str(self.sampID))))
}
})
@ -107,7 +104,7 @@ class SettingWindow(QMainWindow):
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.DETECT_RPEAK_INPUT_ECG_FILENAME +
Path(ConfigParams.ECG_FILTER +
str(self.ui.spinBox_input_freq.value()) +
ConfigParams.ENDSWITH_TXT))))
@ -261,7 +258,7 @@ class MainWindow_detect_Rpeak(QMainWindow):
# TODO：获取检测方法的解耦
method_list = get_method()
if len(method_list) == 0 or method_list is None:
result = Result().failure(info=Constants.DETECT_RPEAK_LOAD_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Method_Not_Exist"])
result = Result().failure(info=Constants.DETECT_RPEAK_LOAD_FAILURE + Constants.FAILURE_REASON["Method_Not_Exist"])
else:
result = Result().success(info=Constants.DETECT_RPEAK_LOAD_FINISHED)
if not result.status:
@ -291,7 +288,7 @@ class MainWindow_detect_Rpeak(QMainWindow):
PublicFunc.__disableAllButton__(self, ButtonState)
# 数据预处理
PublicFunc.progressbar_update(self, 1, 3, Constants.DETECT_RPEAK_PROCESSING_DATA, 0)
PublicFunc.progressbar_update(self, 1, 3, Constants.PREPROCESSING_DATA, 0)
result = self.data.preprocess()
if not result.status:
PublicFunc.text_output(self.ui, "(1/3)" + result.info, Constants.TIPS_TYPE_ERROR)
@ -390,21 +387,35 @@ class Data:
self.RRIV = None
def open_file(self):
if Path(Config["Path"]["Input"]).is_file():
Config["Path"]["Input"] = str(Path(Config["Path"]["Input"]).parent)
if not Path(Config["Path"]["Input"]).exists():
return Result().failure(info=Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Data_Path_Not_Exist"])
return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Path_Not_Exist"])
result = PublicFunc.examine_file(Config["Path"]["Input"], ConfigParams.ECG_FILTER)
if result.status:
Config["Path"]["Input"] = result.data["path"]
Config["InputConfig"]["Freq"] = result.data["freq"]
Config["Path"]["Save"] = str(
Path(Config["Path"]["Save"]) / Path(ConfigParams.RPEAK_FINAL + str(Config["InputConfig"]["Freq"]) + ConfigParams.ENDSWITH_TXT))
else:
return result
try:
self.raw_data = read_csv(Config["Path"]["Input"],
encoding=ConfigParams.UTF8_ENCODING,
header=None).to_numpy().reshape(-1)
except Exception as e:
return Result().failure(info=Constants.INPUT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Read_Data_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.INPUT_FAILURE +
Constants.FAILURE_REASON["Read_Data_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.INPUT_FINISHED)
def preprocess(self):
if self.raw_data is None:
return Result().failure(info=Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Raw_Data_Not_Exist"])
return Result().failure(info=Constants.PREPROCESS_FAILURE +
Constants.FAILURE_REASON["Data_Not_Exist"])
try:
self.processed_data = preprocess(self.raw_data,
@ -412,13 +423,15 @@ class Data:
Config["Filter"]["BandPassLow"],
Config["Filter"]["BandPassHigh"])
except Exception as e:
return Result().failure(info=Constants.DETECT_RPEAK_PROCESS_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Filter_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PREPROCESS_FAILURE +
Constants.FAILURE_REASON["Preprocess_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.DETECT_RPEAK_PROCESS_FINISHED)
return Result().success(info=Constants.PREPROCESS_FINISHED)
def predict_Rpeak(self):
if self.processed_data is None:
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Processed_Data_Not_Exist"])
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE +
Constants.FAILURE_REASON["Data_Not_Exist"])
try:
self.peak, self.interval, self.RRIV = Rpeak_Detection(self.processed_data,
@ -426,18 +439,20 @@ class Data:
Config["PeaksValue"],
Config["DetectMethod"])
except Exception as e:
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Predict_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.DETECT_RPEAK_PREDICT_FAILURE +
Constants.FAILURE_REASON["Predict_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.DETECT_RPEAK_PREDICT_FINISHED)
def save(self, chunk):
if self.peak is None:
return Result().failure(info=Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Peak_Not_Exist"])
return Result().failure(info=Constants.SAVING_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.SAVING_FAILURE + Constants.DETECT_RPEAK_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.SAVING_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.SAVING_FINISHED)