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

优化<数据预处理>的代码

Browse Source 
优化<BCG的J峰算法定位>的代码
This commit is contained in:
Yorusora
2025-05-20 17:05:00 +08:00
parent fada84e115
commit ec16546f47
5 changed files with 106 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
func/Module_detect_Jpeak.py
View File

@ -71,12 +71,9 @@ class SettingWindow(QMainWindow):
Config.update({
"Path": {
"Input": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.DETECT_JPEAK_INPUT_BCG_FILENAME +
str(Config["InputConfig"]["Freq"]) +
ConfigParams.ENDSWITH_TXT))),
Path(str(self.sampID)))),
"Save": str((Path(self.root_path) / ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) / Path(ConfigParams.DETECT_JPEAK_SAVE_FILENAME +
ConfigParams.ENDSWITH_TXT)))
Path(str(self.sampID))))
}
})
@ -110,7 +107,7 @@ class SettingWindow(QMainWindow):
str((Path(self.root_path) /
ConfigParams.PUBLIC_PATH_ORGBCG_TEXT /
Path(str(self.sampID)) /
Path(ConfigParams.DETECT_JPEAK_INPUT_BCG_FILENAME +
Path(ConfigParams.BCG_FILTER +
str(self.ui.spinBox_input_freq.value()) +
ConfigParams.ENDSWITH_TXT))))
@ -295,7 +292,7 @@ class MainWindow_detect_Jpeak(QMainWindow):
PublicFunc.__disableAllButton__(self, ButtonState)
# 数据预处理
PublicFunc.progressbar_update(self, 1, 3, Constants.DETECT_JPEAK_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.interval = 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_JPEAK_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.BCG_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.JPEAK_REVISE + 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_JPEAK_FAILURE_REASON["Read_Data_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.INPUT_FAILURE +
Constants.FAILURE_REASON["Open_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_JPEAK_PROCESS_FAILURE + Constants.DETECT_JPEAK_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,
@ -413,16 +424,19 @@ class Data:
Config["Filter"]["BandPassHigh"],
Config["AmpValue"])
except Exception as e:
return Result().failure(info=Constants.DETECT_JPEAK_PROCESS_FAILURE + Constants.DETECT_JPEAK_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_JPEAK_PROCESS_FINISHED)
return Result().success(info=Constants.PREPROCESS_FINISHED)
def predict_Jpeak(self, model):
if not (Path(model.model_folder_path) / Path(model.selected_model)).exists():
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Model_File_Not_Exist"])
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE +
Constants.FAILURE_REASON["Model_File_Not_Exist"])
if self.processed_data is None:
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Processed_Data_Not_Exist"])
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE +
Constants.FAILURE_REASON["Data_Not_Exist"])
try:
self.peak, self.interval = Jpeak_Detection(model.selected_model,
@ -434,18 +448,20 @@ class Data:
Config["PeaksValue"],
Config["UseCPU"])
except Exception as e:
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Predict_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.DETECT_JPEAK_PREDICT_FAILURE +
Constants.FAILURE_REASON["Predict_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.DETECT_JPEAK_PREDICT_FINISHED)
def save(self, chunk):
if self.peak is None:
return Result().failure(info=Constants.SAVING_FAILURE + Constants.DETECT_JPEAK_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_JPEAK_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)
@ -460,13 +476,13 @@ class Model:
def seek_model(self):
if not Path(Config["ModelFolderPath"]).exists():
return Result().failure(info=Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Model_Path_Not_Exist"])
return Result().failure(info=Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.FAILURE_REASON["Model_Path_Not_Exist"])
try:
self.model_list = [file.name for file in Path(Config["ModelFolderPath"]).iterdir() if file.is_file()]
if len(self.model_list) == 0:
return Result().failure(info=Constants.DETECT_JPEAK_FAILURE_REASON["Model_File_Not_Exist"])
return Result().failure(info=Constants.FAILURE_REASON["Model_File_Not_Exist"])
except Exception as e:
return Result().failure(info=Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.DETECT_JPEAK_FAILURE_REASON["Read_Model_Exception"] + "\n" + format_exc())
return Result().failure(info=Constants.DETECT_JPEAK_LOAD_FAILURE + Constants.FAILURE_REASON["Read_Model_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.DETECT_JPEAK_LOAD_FINISHED)