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-06-02 20:19:40 +08:00
parent 99d355be6b
commit e200c2a488
5 changed files with 309 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
func/Module_approximately_align.py
View File

@ -12,7 +12,7 @@ from overrides import overrides
from pandas import read_csv, DataFrame
from scipy.signal import find_peaks, resample, butter, sosfiltfilt, correlate
from yaml import dump, load, FullLoader
from sklearn.linear_model import TheilSenRegressor
from func.utils.PublicFunc import PublicFunc
from func.utils.Constants import Constants, ConfigParams
from func.utils.Result import Result
@ -44,7 +44,9 @@ ButtonState = {
"radioButton_PABD": False,
"radioButton_NTHO": False,
"radioButton_NABD": False,
"radioButton_custom": False
"radioButton_custom": False,
"radioButton_freqTHO": False,
"radioButton_freqABD": False,
},
"Current": {
"pushButton_input_setting": True,
@ -65,7 +67,9 @@ ButtonState = {
"radioButton_PABD": False,
"radioButton_NTHO": False,
"radioButton_NABD": False,
"radioButton_custom": False
"radioButton_custom": False,
"radioButton_freqTHO": False,
"radioButton_freqABD": False,
}
}
@ -259,6 +263,8 @@ class MainWindow_approximately_align(QMainWindow):
self.ui.radioButton_PABD.clicked.connect(self.__enableAlign__)
self.ui.radioButton_NABD.clicked.connect(self.__enableAlign__)
self.ui.radioButton_custom.clicked.connect(self.__enableAlign__)
self.ui.radioButton_freqTHO.clicked.connect(self.__EstimateFrequencySelect__)
self.ui.radioButton_freqABD.clicked.connect(self.__EstimateFrequencySelect__)
@overrides
@ -305,6 +311,11 @@ class MainWindow_approximately_align(QMainWindow):
self.ui.radioButton_NABD.setText("备选4")
self.ui.radioButton_NTHO.setText("备选2")
self.ui.spinBox_SelectEpoch.setMinimum(0)
self.ui.radioButton_freqABD.setChecked(False)
self.ui.radioButton_freqTHO.setChecked(False)
self.ui.radioButton_freqTHO.setText("备选1")
self.ui.radioButton_freqABD.setText("备选2")
def __plot__(self, *args, **kwargs):
sender = self.sender()
@ -328,7 +339,7 @@ class MainWindow_approximately_align(QMainWindow):
elif sender == self.ui.radioButton_custom:
result = self.DrawPicTryAlign()
elif sender == self.ui.pushButton_ChangeView:
result = self.DrawAlignScatter()
result = self.DrawAlignScatter(*args, **kwargs)
elif sender == self.ui.pushButton_JUMP:
result = self.DrawPicByEpoch(*args, **kwargs)
elif sender == self.ui.pushButton_EM1:
@ -570,7 +581,39 @@ class MainWindow_approximately_align(QMainWindow):
# 绘图
PublicFunc.progressbar_update(self, 1, 1, Constants.DRAWING_DATA, 0)
result = self.__plot__()
response = self.data.get_corr_by_epoch()
epoch_min = response.data["epoch_min"]
epoch_max = response.data["epoch_max"]
tho_bias_list = response.data["tho_bias_list"]
abd_bias_list = response.data["abd_bias_list"]
response = self.data.estimate_frequency(tho_bias_list)
tho_y = response.data["estimate_y"]
tho_frequency = response.data["frequency"]
tho_slope = response.data["slope"]
tho_intercept = response.data["intercept"]
response = self.data.estimate_frequency(abd_bias_list)
abd_y = response.data["estimate_y"]
abd_frequency = response.data["frequency"]
abd_slope = response.data["slope"]
abd_intercept = response.data["intercept"]
result = self.__plot__(epoch_min, epoch_max, tho_bias_list, abd_bias_list, tho_y, abd_y,
tho_frequency, abd_frequency)
Config["estimate"] = {}
Config["estimate"]["tho_frequency"] = tho_frequency
Config["estimate"]["tho_slope"] = tho_slope
Config["estimate"]["tho_intercept"] = tho_intercept
Config["estimate"]["abd_frequency"] = abd_frequency
Config["estimate"]["abd_slope"] = abd_slope
Config["estimate"]["abd_intercept"] = abd_intercept
self.ui.radioButton_freqTHO.setText(str(Config["estimate"]["tho_frequency"]))
self.ui.radioButton_freqABD.setText(str(Config["estimate"]["abd_frequency"]))
ButtonState["Current"]["radioButton_freqTHO"] = True
ButtonState["Current"]["radioButton_freqABD"] = True
if not result.status:
PublicFunc.text_output(self.ui, "(1/1)" + result.info, Constants.TIPS_TYPE_ERROR)
PublicFunc.msgbox_output(self, result.info, Constants.MSGBOX_TYPE_ERROR)
@ -599,6 +642,25 @@ class MainWindow_approximately_align(QMainWindow):
PublicFunc.finish_operation(self, ButtonState)
def __EstimateFrequencySelect__(self):
PublicFunc.__disableAllButton__(self, ButtonState)
if self.ui.radioButton_freqTHO.isChecked():
frequency = Config["estimate"]["tho_frequency"]
Config["estimate_freq"] = frequency
Config["estimate_slope"] = Config["estimate"]["tho_slope"]
Config["estimate_intercept"] = Config["estimate"]["tho_intercept"]
elif self.ui.radioButton_freqABD.isChecked():
frequency = Config["estimate"]["abd_frequency"]
Config["estimate_freq"] = frequency
Config["estimate_slope"] = Config["estimate"]["abd_slope"]
Config["estimate_intercept"] = Config["estimate"]["abd_intercept"]
else:
return
ButtonState["Current"]["pushButton_save"] = True
PublicFunc.finish_operation(self, ButtonState)
def __EpochChange__(self):
# 获取当前值
value = self.ui.spinBox_SelectEpoch.value()
@ -682,7 +744,6 @@ class MainWindow_approximately_align(QMainWindow):
ButtonState["Current"]["pushButton_EP1"] = True
ButtonState["Current"]["pushButton_EP10"] = True
ButtonState["Current"]["pushButton_EP100"] = True
ButtonState["Current"]["pushButton_save"] = True
ButtonState["Current"]["pushButton_ChangeView"] = True
PublicFunc.finish_operation(self, ButtonState)
@ -886,27 +947,26 @@ class MainWindow_approximately_align(QMainWindow):
# 返回图片以便存到QPixImage
return Result().success(info=Constants.DRAW_FINISHED)
def DrawAlignScatter(self):
def DrawAlignScatter(self, epoch_min, epoch_max, tho_bias_list, abd_bias_list, tho_y, abd_y,
tho_frequency, abd_frequency):
try:
response = self.data.get_corr_by_epoch()
epoch_min = response.data["epoch_min"]
epoch_max = response.data["epoch_max"]
tho_bias_list = response.data["tho_bias_list"]
abd_bias_list = response.data["abd_bias_list"]
ax1 = self.fig.add_subplot(211)
ax1.scatter(linspace(epoch_min, epoch_max, len(tho_bias_list)), tho_bias_list, alpha=0.2)
ax1.plot(linspace(epoch_min, epoch_max, len(tho_bias_list)), tho_y, color='orange', label=f"THO Frequency: {tho_frequency} Hz")
ax1.axhline(y=0, color='red', linestyle='--', alpha=0.3)
ax1.set_xlabel("Epoch")
ax1.set_ylabel("Tho Bias / s")
ax1.set_title("THO")
ax1.legend()
ax2 = self.fig.add_subplot(212)
ax2.scatter(linspace(epoch_min, epoch_max, len(abd_bias_list)), abd_bias_list, alpha=0.2)
ax2.plot(linspace(epoch_min, epoch_max, len(abd_bias_list)), abd_y, color='orange', label=f"ABD Frequency: {abd_frequency} Hz")
ax2.axhline(y=0, color='red', linestyle='--', alpha=0.3)
ax2.set_xlabel("Epoch")
ax2.set_ylabel("Abd Bias / s")
ax2.set_title("ABD")
ax2.legend()
self.fig.canvas.draw()
return Result().success(info=Constants.DRAW_FINISHED)
@ -979,8 +1039,18 @@ class Data:
try:
pos = Config["pos"]
ApplyFrequency = Config["ApplyFrequency"]
estimate_freq = Config["estimate_freq"]
estimate_slope = Config["estimate_slope"]
estimate_intercept = Config["estimate_intercept"]
# 保存到csv中
df = DataFrame({"pos": [pos], "epoch": [epoch], "ApplyFrequency": [ApplyFrequency]})
df = DataFrame({"pos": [pos],
"epoch": [epoch],
"ApplyFrequency": [ApplyFrequency],
"estimate_freq": [estimate_freq],
"estimate_slope": [estimate_slope],
"estimate_intercept": [estimate_intercept]
})
df.to_csv(Path(Config["Path"]["Save"]), mode="w", header=True, index=False)
except Exception as e:
return Result().failure(info=Constants.SAVE_FAILURE + Constants.FAILURE_REASON[
@ -1221,6 +1291,7 @@ class Data:
epoch_min = response.data["epoch_min"]
epoch_max = response.data["epoch_max"]
epoch_second = ConfigParams.APPROXIMATELY_ALIGN_CONFIG_NEW_CONTENT["Second_PerEpoch"]
temp_freq = ConfigParams.APPROXIMATELY_ALIGN_CONFIG_NEW_CONTENT["TempFrequency"]
window_epoch = ConfigParams.APPROXIMATELY_ALIGN_CONFIG_NEW_CONTENT["CorrByEpoch"]["window_epoch"]
@ -1231,8 +1302,8 @@ class Data:
pos = Config["pos"] * temp_freq // Config["ApplyFrequency"]
for epoch in range(epoch_min, epoch_max):
SP = epoch * 30 * temp_freq
EP = (epoch + window_epoch) * 30 * temp_freq
SP = epoch * epoch_second * temp_freq
EP = (epoch + window_epoch) * epoch_second * temp_freq
tho_seg = self.processed_Tho[SP:EP]
abd_seg = self.processed_Abd[SP:EP]
@ -1260,3 +1331,32 @@ class Data:
"Get_Corr_By_Epoch_Exception"] + "\n" + format_exc())
return Result().success(info=Constants.APPROXIMATELY_EPOCH_GET_FINISHED, data=result)
@staticmethod
def estimate_frequency(bias_list):
try:
epoch_second = ConfigParams.APPROXIMATELY_ALIGN_CONFIG_NEW_CONTENT["Second_PerEpoch"]
temp_freq = ConfigParams.APPROXIMATELY_ALIGN_CONFIG_NEW_CONTENT["TempFrequency"]
# 生成线性数据
X = linspace(0, len(bias_list), len(bias_list)).reshape(-1, 1)
y = bias_list
theilsen = TheilSenRegressor()
theilsen.fit(X, y)
slope = theilsen.coef_[0]
frequency = 1 - slope / epoch_second / temp_freq if slope != 0 else float('inf')
theilsen_y = theilsen.predict(X)
return Result().success(info=Constants.APPROXIMATELY_ESTIMATE_FREQUENCY_FINISHED,
data={"estimate_y": theilsen_y,
"frequency": frequency,
"slope": slope,
"intercept": theilsen.intercept_},
)
except Exception as e:
return Result().failure(
info=Constants.APPROXIMATELY_ESTIMATE_FREQUENCY_FAILURE + Constants.FAILURE_REASON[
"Estimate_Frequency_Exception"] + "\n" + format_exc())

1
func/utils/ConfigParams.py
View File

@ -85,6 +85,7 @@ class ConfigParams:
"BandPassHigh": 0.7
},
"Multiple_Factor":100,
"Second_PerEpoch": 30,
"CorrByEpoch":
{
"window_epoch": 6

4
func/utils/Constants.py
View File

@ -182,6 +182,10 @@ class Constants:
APPROXIMATELY_EPOCH_GET_FINISHED: str = "获取epoch完成"
APPROXIMATELY_EPOCH_GET_FAILURE: str = "获取epoch失败"
APPROXIMATELY_ESTIMATE_FREQUENCY_CALCULATING: str = "正在估计采样率"
APPROXIMATELY_ESTIMATE_FREQUENCY_FINISHED: str = "估计采样率完成"
APPROXIMATELY_ESTIMATE_FREQUENCY_FAILURE: str = "估计采样率失败"
# 预处理
PREPROCESS_PLOT_LABEL_ORIGINAL_DATA: str = "Original_Data"
PREPROCESS_PLOT_LABEL_PROCESSED_DATA: str = "Processed_Data"