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:
Yorusora
2025-05-20 20:25:45 +08:00
parent d2e0a61056
commit 900912090a
8 changed files with 386 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
func/Module_precisely_align.py
View File

@ -94,6 +94,8 @@ class SettingWindow(QMainWindow):
self.config = None
self.__read_config__()
self.ui.spinBox_input_freq_orgBcg.valueChanged.connect(self.__update_ui__)
self.ui.spinBox_input_freq_BCG.valueChanged.connect(self.__update_ui__)
self.ui.spinBox_input_freq_ECG.valueChanged.connect(self.__update_ui__)
self.ui.pushButton_confirm.clicked.connect(self.__write_config__)
self.ui.pushButton_cancel.clicked.connect(self.__rollback_config__)
@ -112,47 +114,29 @@ class SettingWindow(QMainWindow):
Config.update({
"Path": {
"Input_OrgBCG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_INPUT_ORGBCG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Input_BCG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_INPUT_BCG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Input_Jpeak": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_INPUT_JPEAK_FILENAME +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Input_ECG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_INPUT_ECG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Input_Rpeak": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_INPUT_RPEAK_FILENAME +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Input_Approximately_Align": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_APPROXIMATELY_ALIGNINFO_FILENAME +
ConfigParams.ENDSWITH_CSV))),
Path(str(self.sampID)))),
"Save_AlignInfo": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_LABEL /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_ALIGNINFO_FILENAME +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save_OrgBCG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_ORGBCG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save_BCG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_BCG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save_ECG": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_ECG_FILENAME +
str(Config["InputConfig"]["ECGFreq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save_Jpeak": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_JPEAK_FILENAME +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save_Rpeak": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_PSG_ALIGNED /
Path(str(self.sampID)) / Path(ConfigParams.PRECISELY_ALIGN_SAVE_RPEAK_FILENAME +
ConfigParams.ENDSWITH_TXT)))
Path(str(self.sampID))))
},
"Coordinate": {
"BCG_front_1": 0,
@ -198,6 +182,8 @@ class SettingWindow(QMainWindow):
})
# 数据回显
self.ui.spinBox_input_freq_orgBcg.setValue(Config["InputConfig"]["orgBcgFreq"])
self.ui.spinBox_input_freq_BCG.setValue(Config["InputConfig"]["BCGFreq"])
self.ui.spinBox_input_freq_ECG.setValue(Config["InputConfig"]["ECGFreq"])
self.ui.plainTextEdit_file_path_input_orgBcg.setPlainText(Config["Path"]["Input_OrgBCG"])
self.ui.plainTextEdit_file_path_input_BCG.setPlainText(Config["Path"]["Input_BCG"])
@ -214,6 +200,8 @@ class SettingWindow(QMainWindow):
def __write_config__(self):
# 从界面写入配置
Config["InputConfig"]["orgBcgFreq"] = self.ui.spinBox_input_freq_orgBcg.value()
Config["InputConfig"]["BCGFreq"] = self.ui.spinBox_input_freq_BCG.value()
Config["InputConfig"]["ECGFreq"] = self.ui.spinBox_input_freq_ECG.value()
Config["Path"]["Input_OrgBCG"] = self.ui.plainTextEdit_file_path_input_orgBcg.toPlainText()
Config["Path"]["Input_BCG"] = self.ui.plainTextEdit_file_path_input_BCG.toPlainText()
@ -229,6 +217,8 @@ class SettingWindow(QMainWindow):
Config["Path"]["Save_Rpeak"] = self.ui.plainTextEdit_file_path_save_Rpeak.toPlainText()
# 保存配置到文件
self.config["InputConfig"]["orgBcgFreq"] = self.ui.spinBox_input_freq_orgBcg.value()
self.config["InputConfig"]["BCGFreq"] = self.ui.spinBox_input_freq_BCG.value()
self.config["InputConfig"]["ECGFreq"] = self.ui.spinBox_input_freq_ECG.value()
with open(ConfigParams.PRECISELY_ALIGN_CONFIG_FILE_PATH, "w") as f:
@ -244,55 +234,57 @@ class SettingWindow(QMainWindow):
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_INPUT_ORGBCG_FILENAME +
str(self.ui.spinBox_input_freq_ECG.value()) +
Path(ConfigParams.ORGBCG_RAW +
str(self.ui.spinBox_input_freq_orgBcg.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_input_BCG.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_INPUT_BCG_FILENAME +
str(self.ui.spinBox_input_freq_ECG.value()) +
Path(ConfigParams.BCG_FILTER +
str(self.ui.spinBox_input_freq_BCG.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_input_ECG.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_PSG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_INPUT_ECG_FILENAME +
Path(ConfigParams.ECG_FILTER +
str(self.ui.spinBox_input_freq_ECG.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_save_orgBcg.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_SAVE_ORGBCG_FILENAME +
str(self.ui.spinBox_input_freq_ECG.value()) +
Path(ConfigParams.ORGBCG_SYNC +
str(self.ui.spinBox_input_freq_orgBcg.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_save_BCG.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_SAVE_BCG_FILENAME +
str(self.ui.spinBox_input_freq_ECG.value()) +
Path(ConfigParams.BCG_SYNC +
str(self.ui.spinBox_input_freq_BCG.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_save_ECG.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_PSG_ALIGNED /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_SAVE_ECG_FILENAME +
Path(ConfigParams.ECG_SYNC +
str(self.ui.spinBox_input_freq_ECG.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_save_Jpeak.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_ALIGNED /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_SAVE_JPEAK_FILENAME +
Path(ConfigParams.JPEAK_SYNC +
str(self.ui.spinBox_input_freq_BCG.value()) +
ConfigParams.ENDSWITH_TXT))))
self.ui.plainTextEdit_file_path_save_Rpeak.setPlainText(
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_PSG_ALIGNED /
Path(str(self.sampID)) /
Path(ConfigParams.PRECISELY_ALIGN_SAVE_RPEAK_FILENAME +
Path(ConfigParams.RPEAK_SYNC +
str(self.ui.spinBox_input_freq_ECG.value()) +
ConfigParams.ENDSWITH_TXT))))
@ -650,7 +642,7 @@ class MainWindow_precisely_align(QMainWindow):
PublicFunc.text_output(self.ui, "(1/3)" + result.info, Constants.TIPS_TYPE_INFO)
# 处理数据
PublicFunc.progressbar_update(self, 2, 3, Constants.DRAWING_DATA, 50)
PublicFunc.progressbar_update(self, 2, 3, Constants.PRECISELY_ALIGN_PROCESSING_DATA, 50)
result = self.data.data_process_for_calculate_correlation()
if not result.status:
PublicFunc.text_output(self.ui, "(2/3)" + result.info, Constants.TIPS_TYPE_ERROR)
@ -946,7 +938,7 @@ class MainWindow_precisely_align(QMainWindow):
try:
if self.data is not None:
if self.data.Jpeak is None or self.data.Rpeak is None:
PublicFunc.msgbox_output(self, Constants.PRECISELY_ALIGN_FAILURE_REASON["Data_Not_Exist"], Constants.MSGBOX_TYPE_ERROR)
PublicFunc.msgbox_output(self, Constants.FAILURE_REASON["Data_Not_Exist"], Constants.MSGBOX_TYPE_ERROR)
return
sender = self.sender()
@ -1435,11 +1427,69 @@ class Data:
self.argmax_ECG = None
def open_file(self):
if ((not Path(Config["Path"]["Input_BCG"]).exists())
if Path(Config["Path"]["Input_OrgBCG"]).is_file():
Config["Path"]["Input_OrgBCG"] = str(Path(Config["Path"]["Input_OrgBCG"]).parent)
if Path(Config["Path"]["Input_BCG"]).is_file():
Config["Path"]["Input_BCG"] = str(Path(Config["Path"]["Input_BCG"]).parent)
if Path(Config["Path"]["Input_Jpeak"]).is_file():
Config["Path"]["Input_Jpeak"] = str(Path(Config["Path"]["Input_Jpeak"]).parent)
if Path(Config["Path"]["Input_ECG"]).is_file():
Config["Path"]["Input_ECG"] = str(Path(Config["Path"]["Input_ECG"]).parent)
if Path(Config["Path"]["Input_Rpeak"]).is_file():
Config["Path"]["Input_Rpeak"] = str(Path(Config["Path"]["Input_Rpeak"]).parent)
if Path(Config["Path"]["Input_Approximately_Align"]).is_file():
Config["Path"]["Input_Approximately_Align"] = str(Path(Config["Path"]["Input_Approximately_Align"]).parent)
result = PublicFunc.examine_file(Config["Path"]["Input_OrgBCG"], ConfigParams.ORGBCG_RAW)
if result.status:
Config["Path"]["Input_OrgBCG"] = result.data["path"]
Config["InputConfig"]["orgBcgFreq"] = result.data["freq"]
Config["Path"]["Input_Approximately_Align"] = str(
Path(Config["Path"]["Input_Approximately_Align"]) / Path(
ConfigParams.APPROXIMATELY_ALIGN_INFO + ConfigParams.ENDSWITH_CSV))
Config["Path"]["Save_AlignInfo"] = str(
Path(Config["Path"]["Save_AlignInfo"]) / Path(
ConfigParams.PRECISELY_ALIGN_INFO + ConfigParams.ENDSWITH_TXT))
Config["Path"]["Save_OrgBCG"] = str(
Path(Config["Path"]["Save_OrgBCG"]) / Path(
ConfigParams.ORGBCG_SYNC + str(Config["InputConfig"]["orgBcgFreq"]) + ConfigParams.ENDSWITH_TXT))
else:
return result
result = PublicFunc.examine_file(Config["Path"]["Input_BCG"], ConfigParams.BCG_FILTER)
if result.status:
Config["Path"]["Input_BCG"] = result.data["path"]
Config["InputConfig"]["BCGFreq"] = result.data["freq"]
Config["Path"]["Input_Jpeak"] = str(
Path(Config["Path"]["Input_Jpeak"]) / Path(ConfigParams.JPEAK_REVISE_CORRECTED + str(Config["InputConfig"]["BCGFreq"]) + ConfigParams.ENDSWITH_TXT))
Config["Path"]["Save_BCG"] = str(
Path(Config["Path"]["Save_BCG"]) / Path(
ConfigParams.BCG_SYNC + str(Config["InputConfig"]["BCGFreq"]) + ConfigParams.ENDSWITH_TXT))
Config["Path"]["Save_Jpeak"] = str(
Path(Config["Path"]["Save_Jpeak"]) / Path(
ConfigParams.JPEAK_SYNC + str(Config["InputConfig"]["BCGFreq"]) + ConfigParams.ENDSWITH_TXT))
else:
return result
result = PublicFunc.examine_file(Config["Path"]["Input_ECG"], ConfigParams.ECG_FILTER)
if result.status:
Config["Path"]["Input_ECG"] = result.data["path"]
Config["InputConfig"]["ECGFreq"] = result.data["freq"]
Config["Path"]["Input_Rpeak"] = str(
Path(Config["Path"]["Input_Rpeak"]) / Path(ConfigParams.RPEAK_FINAL_CORRECTED + str(Config["InputConfig"]["ECGFreq"]) + ConfigParams.ENDSWITH_TXT))
Config["Path"]["Save_ECG"] = str(
Path(Config["Path"]["Save_ECG"]) / Path(
ConfigParams.ECG_SYNC + str(Config["InputConfig"]["ECGFreq"]) + ConfigParams.ENDSWITH_TXT))
Config["Path"]["Save_Rpeak"] = str(
Path(Config["Path"]["Save_Rpeak"]) / Path(
ConfigParams.RPEAK_SYNC + str(Config["InputConfig"]["ECGFreq"]) + ConfigParams.ENDSWITH_TXT))
else:
return result
if ((not Path(Config["Path"]["Input_OrgBCG"]).exists())
or (not Path(Config["Path"]["Input_BCG"]).exists())
or (not Path(Config["Path"]["Input_Jpeak"]).exists())
or (not Path(Config["Path"]["Input_ECG"]).exists())
or (not Path(Config["Path"]["Input_Rpeak"]).exists())):
return Result().failure(info=Constants.INPUT_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Data_Path_Not_Exist"])
return Result().failure(info=Constants.INPUT_FAILURE + Constants.FAILURE_REASON["Data_Not_Exist"])
try:
self.raw_orgBcg = read_csv(Config["Path"]["Input_OrgBCG"],
@ -1460,13 +1510,14 @@ class Data:
self.argmax_BCG = np_argmax(self.raw_BCG)
self.argmax_ECG = np_argmax(self.raw_ECG)
except Exception as e:
return Result().failure(info=Constants.INPUT_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Read_Data_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.INPUT_FAILURE +
Constants.FAILURE_REASON["Open_Data_Exception"] + "\n" + format_exc())
try:
df = read_csv(Config["Path"]["Input_Approximately_Align"])
pos = df["pos"].values[-1]
ApplyFrequency = df["ApplyFrequency"].values[-1]
self.approximately_align_pos = int(pos * (Config["InputConfig"]["ECGFreq"] / ApplyFrequency))
self.approximately_align_pos = int(pos * (Config["InputConfig"]["UseFreq"] / ApplyFrequency))
if self.approximately_align_pos > 0:
self.BCG_early = False
elif self.approximately_align_pos < 0:
@ -1481,7 +1532,8 @@ class Data:
def data_process_for_calculate_correlation(self):
if self.Jpeak is None or self.Rpeak is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_PROCESS_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Data_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_PROCESS_FAILURE +
Constants.FAILURE_REASON["Data_Not_Exist"])
try:
self.JJIs = diff(self.Jpeak)
@ -1491,18 +1543,21 @@ class Data:
result = {"JJIVs": JJIVs, "RRIVs": RRIVs}
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_PROCESS_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Process_Data_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_PROCESS_FAILURE +
Constants.FAILURE_REASON["Process_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_PROCESS_FINISHED, data=result)
def calculate_correlation_front(self, mode, shift_front=None):
if ((Config["IV_Coordinate"]["BCG_front_1"] == Config["IV_Coordinate"]["BCG_front_2"])
or (Config["IV_Coordinate"]["ECG_front_1"] == Config["IV_Coordinate"]["ECG_front_2"])):
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_Value_Equal"])
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT +
Constants.FAILURE_REASON["Calculate_Correlation_Value_Equal"])
if ((Config["IV_Coordinate"]["BCG_front_2"] - Config["IV_Coordinate"]["BCG_front_1"])
>= (Config["IV_Coordinate"]["ECG_front_2"] - Config["IV_Coordinate"]["ECG_front_1"])):
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_JJIVRange_too_Large"])
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT +
Constants.FAILURE_REASON["Calculate_Correlation_JJIVRange_too_Large"])
try:
if mode == "init":
@ -1531,7 +1586,7 @@ class Data:
offset_interval = np_sum(RRI[:shift]) - (Jpeak_cut[0] - Rpeak_cut[0])
anchor_J = self.Jpeak[Config["IV_Coordinate"]["BCG_front_1"]]
anchor_R = self.Rpeak[Config["IV_Coordinate"]["ECG_front_1"] + shift]
offset_interval_duration = offset_interval / Config["InputConfig"]["ECGFreq"]
offset_interval_duration = offset_interval / Config["InputConfig"]["UseFreq"]
Config["front"]["shift"] = shift
Config["front"]["offset_interval"] = offset_interval
@ -1556,18 +1611,21 @@ class Data:
}
}
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_FRONT +
Constants.FAILURE_REASON["Calculate_Correlation_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_CALCULATE_FINISHED_FRONT, data=result)
def calculate_correlation_back(self, mode, shift_back=None):
if ((Config["IV_Coordinate"]["BCG_back_1"] == Config["IV_Coordinate"]["BCG_back_2"])
or (Config["IV_Coordinate"]["ECG_back_1"] == Config["IV_Coordinate"]["ECG_back_2"])):
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_Value_Equal"])
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK +
Constants.FAILURE_REASON["Calculate_Correlation_Value_Equal"])
if ((Config["IV_Coordinate"]["BCG_back_2"] - Config["IV_Coordinate"]["BCG_back_1"])
>= (Config["IV_Coordinate"]["ECG_back_2"] - Config["IV_Coordinate"]["ECG_back_1"])):
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_JJIVRange_too_Large"])
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK +
Constants.FAILURE_REASON["Calculate_Correlation_JJIVRange_too_Large"])
try:
if mode == "init":
@ -1596,7 +1654,7 @@ class Data:
offset_interval = np_sum(RRI[:shift]) - (Jpeak_cut[0] - Rpeak_cut[0])
anchor_J = self.Jpeak[Config["IV_Coordinate"]["BCG_back_1"]]
anchor_R = self.Rpeak[Config["IV_Coordinate"]["ECG_back_1"] + shift]
offset_interval_duration = offset_interval / Config["InputConfig"]["ECGFreq"]
offset_interval_duration = offset_interval / Config["InputConfig"]["UseFreq"]
Config["back"]["shift"] = shift
Config["back"]["offset_interval"] = offset_interval
@ -1621,7 +1679,8 @@ class Data:
}
}
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK + Constants.PRECISELY_ALIGN_FAILURE_REASON["Calculate_Correlation_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_CALCULATE_FAILURE_BACK +
Constants.FAILURE_REASON["Calculate_Correlation_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_CALCULATE_FINISHED_BACK, data=result)
@ -1630,7 +1689,7 @@ class Data:
if mode == "init":
anchor0 = [Config["front"]["anchor_R"], Config["front"]["anchor_J"]]
anchor1 = [Config["back"]["anchor_R"], Config["back"]["anchor_J"]]
Config["orgfs"] = ((int(anchor1[1]) - int(anchor0[1])) * Config["InputConfig"]["ECGFreq"] /
Config["orgfs"] = ((int(anchor1[1]) - int(anchor0[1])) * Config["InputConfig"]["UseFreq"] /
(int(anchor1[0]) - int(anchor0[0])))
Config["offset_anchor"] = anchor0[0] - anchor0[1]
@ -1654,11 +1713,11 @@ class Data:
anchor0[1] = anchor0[1] + off
anchor1[1] = anchor1[1] + off
self.res_BCG = resample(self.res_BCG, orgfs, Config["InputConfig"]["ECGFreq"])
self.res_orgBcg = resample(self.res_orgBcg, orgfs, Config["InputConfig"]["ECGFreq"])
self.res_BCG = resample(self.res_BCG, orgfs, Config["InputConfig"]["UseFreq"])
self.res_orgBcg = resample(self.res_orgBcg, orgfs, Config["InputConfig"]["UseFreq"])
anchor0[1] = round(int(anchor0[1]) * Config["InputConfig"]["ECGFreq"] / orgfs)
anchor1[1] = round(int(anchor1[1]) * Config["InputConfig"]["ECGFreq"] / orgfs)
anchor0[1] = round(int(anchor0[1]) * Config["InputConfig"]["UseFreq"] / orgfs)
anchor1[1] = round(int(anchor1[1]) * Config["InputConfig"]["UseFreq"] / orgfs)
off = anchor1[0] - anchor1[1]
if off > 0:
@ -1705,7 +1764,8 @@ class Data:
else:
raise ValueError("模式不存在")
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_ALIGN_CORRELATION_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Correlation_Align_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_ALIGN_CORRELATION_FAILURE +
Constants.FAILURE_REASON["Correlation_Align_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_ALIGN_CORRELATION_FINISHED, data=result)
@ -1750,8 +1810,8 @@ class Data:
self.cut_Jpeak.append(peaks[idx])
self.cut_Jpeak = asarray(self.cut_Jpeak).astype(int)
frontcut_index_BCG = int((self.argmax_BCG - np_argmax(self.res_BCG) / Config["InputConfig"]["ECGFreq"] * Config["orgfs"]))
backcut_index_BCG = int(len(self.res_BCG) / Config["InputConfig"]["ECGFreq"] * Config["orgfs"] + np_argmax(self.raw_BCG) - np_argmax(self.res_BCG) / Config["InputConfig"]["ECGFreq"] * Config["orgfs"])
frontcut_index_BCG = int((self.argmax_BCG - np_argmax(self.res_BCG) / Config["InputConfig"]["UseFreq"] * Config["orgfs"]))
backcut_index_BCG = int(len(self.res_BCG) / Config["InputConfig"]["UseFreq"] * Config["orgfs"] + np_argmax(self.raw_BCG) - np_argmax(self.res_BCG) / Config["InputConfig"]["UseFreq"] * Config["orgfs"])
frontcut_index_ECG = self.argmax_ECG - np_argmax(self.cut_ECG)
backcut_index_ECG = len(self.cut_ECG) + self.argmax_ECG - np_argmax(self.cut_ECG)
@ -1760,7 +1820,8 @@ class Data:
Config["frontcut_index_ECG"] = frontcut_index_ECG
Config["backcut_index_ECG"] = backcut_index_ECG
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_POSTPROCESS_VIEW_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["PostProcess_Align_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_POSTPROCESS_VIEW_FAILURE +
Constants.FAILURE_REASON["PostProcess_Align_Exception"] + "\n" + format_exc())
info = f"{Constants.PRECISELY_ALIGN_POSTPROCESS_VIEW_FINISHED}，BCG前后段被切割的坐标值为[{frontcut_index_BCG}, {backcut_index_BCG}]，ECG前后段被切割的坐标值为[{frontcut_index_ECG}, {backcut_index_ECG}]"
return Result().success(info=info)
@ -1800,62 +1861,73 @@ class Data:
save_data = [str(save_data)]
DataFrame(save_data).to_csv(Config["Path"]["Save_AlignInfo"], index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_ALIGNINFO_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_ALIGNINFO_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_ALIGNINFO_FINISHED)
def save_res_orgBcg(self, chunk):
if self.res_orgBcg is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_ORGBCG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["res_orgBcg_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_ORGBCG_FAILURE +
Constants.FAILURE_REASON["res_orgBcg_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save_OrgBCG"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_ORGBCG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_ORGBCG_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_RES_ORGBCG_FINISHED)
def save_res_BCG(self, chunk):
if self.res_BCG is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_BCG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["res_BCG_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_BCG_FAILURE +
Constants.FAILURE_REASON["res_BCG_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save_BCG"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_BCG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_RES_BCG_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_RES_BCG_FINISHED)
def save_cut_ECG(self, chunk):
if self.cut_ECG is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_ECG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["cut_ECG_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_ECG_FAILURE +
Constants.FAILURE_REASON["cut_ECG_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save_ECG"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_ECG_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_ECG_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_CUT_ECG_FINISHED)
def save_Jpeak(self, chunk):
if self.cut_Jpeak is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_JPEAK_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["cut_Jpeak_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_JPEAK_FAILURE +
Constants.FAILURE_REASON["cut_Jpeak_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save_Jpeak"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_JPEAK_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_JPEAK_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_CUT_JPEAK_FINISHED)
def save_Rpeak(self, chunk):
if self.cut_Rpeak is None:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_RPEAK_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["cut_Rpeak_Not_Exist"])
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_RPEAK_FAILURE +
Constants.FAILURE_REASON["cut_Rpeak_Not_Exist"])
try:
chunk.to_csv(Config["Path"]["Save_Rpeak"], mode='a', index=False, header=False)
except Exception as e:
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_RPEAK_FAILURE + Constants.PRECISELY_ALIGN_FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.PRECISELY_ALIGN_SAVING_CUT_RPEAK_FAILURE +
Constants.FAILURE_REASON["Save_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.PRECISELY_ALIGN_SAVING_CUT_RPEAK_FINISHED)