@ -3,12 +3,15 @@ from pathlib import Path
from traceback import format_exc
import matplotlib . pyplot as plt
from PySide6 . QtWidgets import QMessageBox , QMainWindow , QApplication
from PySide6 . QtGui import QColor
from PySide6 . QtWidgets import QMessageBox , QMainWindow , QApplication , QTableWidget , QTableWidgetItem
from matplotlib import gridspec
from matplotlib . backends . backend_qt import NavigationToolbar2QT
from matplotlib . backends . backend_qtagg import FigureCanvasQTAgg
from numpy import array , int64 , append , zeros , full , where , nan , arange , float64 , isnan , place , count_nonzero , all as np_all
from overrides import overrides
from pandas import read_csv , DataFrame
from scipy . signal import resample , iirfilter , lfilter
from yaml import dump , load , FullLoader
from func . utils . PublicFunc import PublicFunc
@ -68,6 +71,80 @@ class SettingWindow(QMainWindow):
self . sampID = sampID
self . config = None
self . __read_config__ ( )
self . ui . spinBox_input_freq_signal_BCG . valueChanged . connect ( self . __update_ui__ )
self . ui . radioButton_30s_mode . clicked . connect ( self . __update_mode__ )
self . ui . radioButton_10s_mode . clicked . connect ( self . __update_mode__ )
self . ui . pushButton_confirm . clicked . connect ( self . __write_config__ )
self . ui . pushButton_cancel . clicked . connect ( self . __rollback_config__ )
self . ui . pushButton_cancel . clicked . connect ( self . close )
def __read_config__ ( self ) :
if not Path ( ConfigParams . BCG_QUALITY_LABEL_CONFIG_FILE_PATH ) . exists ( ) :
with open ( ConfigParams . BCG_QUALITY_LABEL_CONFIG_FILE_PATH , " w " ) as f :
dump ( ConfigParams . BCG_QUALITY_LABEL_CONFIG_NEW_CONTENT , f )
with open ( ConfigParams . BCG_QUALITY_LABEL_CONFIG_FILE_PATH , " r " ) as f :
file_config = load ( f . read ( ) , Loader = FullLoader )
Config . update ( file_config )
self . config = file_config
Config . update ( {
" Path " : {
" Input_BCG " : str ( ( Path ( self . root_path ) / ConfigParams . PUBLIC_PATH_ORGBCG_ALIGNED /
Path ( str ( self . sampID ) ) ) ) ,
" Input_Artifact " : str ( ( Path ( self . root_path ) / ConfigParams . PUBLIC_PATH_LABEL /
Path ( str ( self . sampID ) ) ) ) ,
" Save " : str ( ( Path ( self . root_path ) / ConfigParams . PUBLIC_PATH_LABEL /
Path ( str ( self . sampID ) ) ) ) ,
} ,
" Mode " : " Undefined " ,
" DataPartNum " : 0 ,
" CurrentDataIdx " : 0 ,
" CurrentPartNum " : 1 ,
" LongestContinuousTime " : 0 ,
" ArtifactTimePercentage " : 0
} )
# 数据回显
self . ui . spinBox_input_freq_signal_BCG . setValue ( Config [ " InputConfig " ] [ " BCGFreq " ] )
self . ui . plainTextEdit_file_path_input_signal_BCG . setPlainText ( Config [ " Path " ] [ " Input_BCG " ] )
self . ui . plainTextEdit_file_path_input_artifact . setPlainText ( Config [ " Path " ] [ " Input_Artifact " ] )
self . ui . plainTextEdit_file_path_save . setPlainText ( Config [ " Path " ] [ " Save " ] )
def __write_config__ ( self ) :
# 从界面写入配置
Config [ " InputConfig " ] [ " BCGFreq " ] = self . ui . spinBox_input_freq_signal_BCG . value ( )
Config [ " Path " ] [ " Input_BCG " ] = self . ui . plainTextEdit_file_path_input_signal_BCG . toPlainText ( )
Config [ " Path " ] [ " Input_Artifact " ] = self . ui . plainTextEdit_file_path_input_artifact . toPlainText ( )
Config [ " Path " ] [ " Save " ] = self . ui . plainTextEdit_file_path_save . toPlainText ( )
# 保存配置到文件
self . config [ " InputConfig " ] [ " BCGFreq " ] = self . ui . spinBox_input_freq_signal_BCG . value ( )
with open ( ConfigParams . BCG_QUALITY_LABEL_CONFIG_FILE_PATH , " w " ) as f :
dump ( self . config , f )
self . close ( )
def __rollback_config__ ( self ) :
self . __read_config__ ( )
def __update_ui__ ( self ) :
self . ui . plainTextEdit_file_path_input_signal_BCG . setPlainText (
str ( ( Path ( self . root_path ) /
ConfigParams . PUBLIC_PATH_ORGBCG_ALIGNED /
Path ( str ( self . sampID ) ) /
Path ( ConfigParams . BCG_SYNC +
str ( self . ui . spinBox_input_freq_signal_BCG . value ( ) ) +
ConfigParams . ENDSWITH_TXT ) ) ) )
def __update_mode__ ( self ) :
if self . ui . radioButton_30s_mode . isChecked ( ) :
Config [ " Mode " ] = " 30s "
elif self . ui . radioButton_10s_mode . isChecked ( ) :
Config [ " Mode " ] = " 10s "
class MainWindow_bcg_quality_label ( QMainWindow ) :
@ -86,4 +163,589 @@ class MainWindow_bcg_quality_label(QMainWindow):
# 初始化进度条
self . progressbar = None
PublicFunc . add_progressbar ( self )
PublicFunc . add_progressbar ( self )
#初始化画框
self . fig = None
self . canvas = None
self . figToolbar = None
self . gs = None
self . ax0 = None
self . line_data = None
self . msgBox = QMessageBox ( )
self . msgBox . setWindowTitle ( Constants . MAINWINDOW_MSGBOX_TITLE )
@overrides
def show ( self , root_path , sampID ) :
super ( ) . show ( )
self . root_path = root_path
self . sampID = sampID
self . setting = SettingWindow ( root_path , sampID )
# 初始化画框
self . fig = plt . figure ( figsize = ( 12 , 9 ) , dpi = 100 )
self . canvas = FigureCanvasQTAgg ( self . fig )
self . figToolbar = NavigationToolbar2QT ( self . canvas )
for action in self . figToolbar . _actions . values ( ) :
action . setEnabled ( False )
for action in self . figToolbar . actions ( ) :
if action . text ( ) == " Subplots " or action . text ( ) == " Customize " :
self . figToolbar . removeAction ( action )
self . ui . verticalLayout_canvas . addWidget ( self . canvas )
self . ui . verticalLayout_canvas . addWidget ( self . figToolbar )
self . gs = gridspec . GridSpec ( 1 , 1 , height_ratios = [ 1 ] )
self . fig . subplots_adjust ( top = 0.98 , bottom = 0.05 , right = 0.98 , left = 0.1 , hspace = 0 , wspace = 0 )
self . ax0 = self . fig . add_subplot ( self . gs [ 0 ] )
self . ax0 . grid ( True )
self . ax0 . xaxis . set_major_formatter ( ConfigParams . FORMATTER )
PublicFunc . __resetAllButton__ ( self , ButtonState )
self . ui . tableWidget_a1 . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_a2 . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_b1 . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_b2 . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_c . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_f . setEditTriggers ( QTableWidget . NoEditTriggers )
self . ui . tableWidget_a1 . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_a2 . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_b1 . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_b2 . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_c . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_f . horizontalHeader ( ) . setStretchLastSection ( True )
self . ui . tableWidget_a1 . setHorizontalHeaderLabels ( [ ' a1 ' ] )
self . ui . tableWidget_a2 . setHorizontalHeaderLabels ( [ ' a2 ' ] )
self . ui . tableWidget_b1 . setHorizontalHeaderLabels ( [ ' b1 ' ] )
self . ui . tableWidget_b2 . setHorizontalHeaderLabels ( [ ' b2 ' ] )
self . ui . tableWidget_c . setHorizontalHeaderLabels ( [ ' c ' ] )
self . ui . tableWidget_f . setHorizontalHeaderLabels ( [ ' None ' ] )
self . ui . pushButton_input_setting . clicked . connect ( self . setting . show )
self . ui . pushButton_input . clicked . connect ( self . __slot_btn_input__ )
@overrides
def closeEvent ( self , event ) :
reply = QMessageBox . question ( self , ' 确认 ' , ' 确认退出吗? ' , QMessageBox . Yes | QMessageBox . No , QMessageBox . No )
if reply == QMessageBox . Yes :
PublicFunc . __disableAllButton__ ( self , ButtonState )
PublicFunc . statusbar_show_msg ( self , PublicFunc . format_status_msg ( Constants . SHUTTING_DOWN ) )
QApplication . processEvents ( )
# 清空画框
if self . ax0 is not None :
self . ax0 . clear ( )
# 释放资源
del self . data
self . fig . clf ( )
plt . close ( self . fig )
self . deleteLater ( )
collect ( )
self . canvas = None
event . accept ( )
else :
event . ignore ( )
def __reset__ ( self ) :
ButtonState [ " Current " ] . update ( ButtonState [ " Default " ] . copy ( ) )
def __plot__ ( self ) :
# 清空画框
self . reset_axes ( )
sender = self . sender ( )
if sender == self . ui . pushButton_input :
try :
artifact_type_seq = array ( [ ] )
artifact_type_seq = artifact_type_seq . astype ( int64 )
if self . ui . checkBox_type1 . isChecked ( ) :
artifact_type_seq = append ( artifact_type_seq , 1 )
if self . ui . checkBox_type2 . isChecked ( ) :
artifact_type_seq = append ( artifact_type_seq , 2 )
if self . ui . checkBox_type3 . isChecked ( ) :
artifact_type_seq = append ( artifact_type_seq , 3 )
if self . ui . checkBox_type4 . isChecked ( ) :
artifact_type_seq = append ( artifact_type_seq , 4 )
if self . ui . checkBox_type5 . isChecked ( ) :
artifact_type_seq = append ( artifact_type_seq , 5 )
# 是否显示去除工频噪声
if self . ui . checkBox_display_afterfilter . isChecked ( ) :
display_data = self . data . BCG_without_industrialFrequencyNoise
else :
display_data = self . data . BCG
if Config [ " Mode " ] == " 30s " :
length = Config [ " InputConfig " ] [ " UseFreq " ] * 30
elif Config [ " Mode " ] == " 10s " :
length = Config [ " InputConfig " ] [ " UseFreq " ] * 10
else :
raise ValueError ( " 模式不存在 " )
# 绘制数据和体动
mask = array ( [ arange ( length ) , arange ( length ) , arange ( length ) , arange ( length ) , arange ( length ) , arange ( length ) ] )
mask = mask . astype ( float64 )
self . ax0 . plot (
arange ( Config [ " CurrentDataIdx " ] , Config [ " CurrentDataIdx " ] + length ) ,
display_data [ Config [ " CurrentDataIdx " ] : Config [ " CurrentDataIdx " ] + length ] ,
label = Constants . BCG_QUALITY_LABEL_PLOT_LABEL_SIGNAL , color = Constants . PLOT_COLOR_BLUE )
for i in artifact_type_seq :
mask [ i ] = self . data . artifact_mask [ Config [ " CurrentDataIdx " ] : Config [ " CurrentDataIdx " ] + length ] == i
mask [ i ] = ( display_data [ Config [ " CurrentDataIdx " ] : Config [ " CurrentDataIdx " ] + length ] *
mask [ i ] ) . astype ( float64 )
place ( mask [ i ] , mask [ i ] == 0 , nan )
self . ax0 . plot ( arange ( Config [ " CurrentDataIdx " ] , Config [ " CurrentDataIdx " ] + length ) , mask [ i ] ,
label = f " { Constants . BCG_QUALITY_LABEL_PLOT_LABEL_ARTIFACT } { i } " , color = Constants . PLOT_COLOR_RED ,
linestyle = " - " )
isArtifact = mask [ ~ np_all ( mask == range ( 0 , length ) , axis = 1 ) ]
isArtifact = ~ isnan ( isArtifact ) . all ( axis = 0 )
# 计算最长连续时间并更新信息
longest_continuous_max_length = 0
longest_continuous_current_length = 0
longest_continuous_end = 0
for index , value in enumerate ( isArtifact ) :
if value == False :
longest_continuous_current_length + = 1
if longest_continuous_current_length > longest_continuous_max_length :
longest_continuous_end = index
longest_continuous_max_length = max ( longest_continuous_max_length ,
longest_continuous_current_length )
else :
longest_continuous_current_length = 0
longest_continuous_start = longest_continuous_end - longest_continuous_max_length + 1
if self . ui . checkBox_highlight_longest_continuous . isChecked ( ) :
self . ax0 . plot (
arange ( Config [ " CurrentDataIdx " ] + longest_continuous_start ,
Config [ " CurrentDataIdx " ] + longest_continuous_end ) ,
display_data [
Config [ " CurrentDataIdx " ] + longest_continuous_start : Config [ " CurrentDataIdx " ] + longest_continuous_end ] ,
label = Constants . BCG_QUALITY_LABEL_PLOT_LABEL_LONGEST_CONTINUOUS ,
color = Constants . PLOT_COLOR_PURPLE , linestyle = " - " , zorder = 3 )
self . ui . lineEdit_longest_continuous_time . setText (
str ( round ( longest_continuous_max_length / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) , 2 ) ) + " 秒 " )
# 计算体动时间占比并更新信息
self . ui . lineEdit_artifact_time_percentage . setText (
str ( round ( 100 * count_nonzero ( isArtifact ) / len ( isArtifact ) , 2 ) ) + " % " )
self . ax0 . legend ( loc = Constants . PLOT_UPPER_RIGHT )
self . canvas . draw ( )
except Exception as e :
return Result ( ) . failure ( info = Constants . DRAW_FAILURE + " \n " + format_exc ( ) )
return Result ( ) . success ( info = Constants . DRAW_FINISHED )
else :
self . canvas . draw ( )
self . ax0 . autoscale ( False )
return Result ( ) . failure ( info = Constants . DRAW_FAILURE )
def __slot_btn_input__ ( self ) :
PublicFunc . __disableAllButton__ ( self , ButtonState )
self . reset_axes ( )
self . canvas . draw ( )
self . data = Data ( )
# 导入数据
PublicFunc . progressbar_update ( self , 1 , 5 , Constants . INPUTTING_DATA , 0 )
result = self . data . open_file ( )
if not result . status :
PublicFunc . text_output ( self . ui , " (1/5) " + result . info , Constants . TIPS_TYPE_ERROR )
PublicFunc . msgbox_output ( self , result . info , Constants . MSGBOX_TYPE_ERROR )
PublicFunc . finish_operation ( self , ButtonState )
return
else :
PublicFunc . text_output ( self . ui , " (1/5) " + result . info , Constants . TIPS_TYPE_INFO )
# 获取存档
PublicFunc . progressbar_update ( self , 2 , 5 , Constants . LOADING_ARCHIVE , 30 )
result = self . data . get_archive ( )
if not result . status :
PublicFunc . text_output ( self . ui , " (2/5) " + result . info , Constants . TIPS_TYPE_ERROR )
PublicFunc . msgbox_output ( self , result . info , Constants . MSGBOX_TYPE_ERROR )
PublicFunc . finish_operation ( self , ButtonState )
return
else :
PublicFunc . text_output ( self . ui , " (2/5) " + result . info , Constants . TIPS_TYPE_INFO )
# 重采样
PublicFunc . progressbar_update ( self , 3 , 5 , Constants . RESAMPLING_DATA , 40 )
result = self . data . resample ( )
if not result . status :
PublicFunc . text_output ( self . ui , " (3/5) " + result . info , Constants . TIPS_TYPE_ERROR )
PublicFunc . msgbox_output ( self , result . info , Constants . MSGBOX_TYPE_ERROR )
PublicFunc . finish_operation ( self , ButtonState )
return
else :
PublicFunc . text_output ( self . ui , " (3/5) " + result . info , Constants . TIPS_TYPE_INFO )
# 数据预处理
PublicFunc . progressbar_update ( self , 4 , 5 , Constants . PREPROCESSING_DATA , 70 )
result = self . data . preprocess ( )
if not result . status :
PublicFunc . text_output ( self . ui , " (4/5) " + result . info , Constants . TIPS_TYPE_ERROR )
PublicFunc . msgbox_output ( self , result . info , Constants . MSGBOX_TYPE_ERROR )
PublicFunc . finish_operation ( self , ButtonState )
return
else :
PublicFunc . text_output ( self . ui , " (4/5) " + result . info , Constants . TIPS_TYPE_INFO )
# 绘图
PublicFunc . progressbar_update ( self , 5 , 5 , Constants . DRAWING_DATA , 90 )
result = self . __plot__ ( )
if not result . status :
PublicFunc . text_output ( self . ui , " (5/5) " + result . info , Constants . TIPS_TYPE_ERROR )
PublicFunc . msgbox_output ( self , result . info , Constants . MSGBOX_TYPE_ERROR )
PublicFunc . finish_operation ( self , ButtonState )
return
else :
PublicFunc . text_output ( self . ui , " (5/5) " + result . info , Constants . TIPS_TYPE_INFO )
self . __reset__ ( )
self . change_tablewidget_mode ( )
self . update_tableWidget ( )
self . update_status ( )
ButtonState [ " Current " ] [ " pushButton_input_setting " ] = True
ButtonState [ " Current " ] [ " pushButton_input " ] = True
ButtonState [ " Current " ] [ " pushButton_invalid_signal_label " ] = True
ButtonState [ " Current " ] [ " pushButton_Ctype_signal_label " ] = True
ButtonState [ " Current " ] [ " pushButton_prev " ] = True
ButtonState [ " Current " ] [ " pushButton_next " ] = True
ButtonState [ " Current " ] [ " pushButton_save " ] = True
ButtonState [ " Current " ] [ " pushButton_a1 " ] = True
ButtonState [ " Current " ] [ " pushButton_a2 " ] = True
ButtonState [ " Current " ] [ " pushButton_b1 " ] = True
ButtonState [ " Current " ] [ " pushButton_b2 " ] = True
ButtonState [ " Current " ] [ " pushButton_c " ] = True
ButtonState [ " Current " ] [ " pushButton_f " ] = True
for action in self . figToolbar . _actions . values ( ) :
action . setEnabled ( True )
PublicFunc . finish_operation ( self , ButtonState )
def update_status ( self ) :
if Config [ " Mode " ] == " 30s " :
hour = int ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 30 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) / / 3600 )
min = int ( ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 30 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) % 3600 ) / / 60 )
sec = int ( ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 30 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) % 3600 ) % 60 )
elif Config [ " Mode " ] == " 10s " :
hour = int ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 10 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) / / 3600 )
min = int ( ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 10 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) % 3600 ) / / 60 )
sec = int ( ( ( ( Config [ " CurrentDataIdx " ] + ( Config [ " InputConfig " ] [ " UseFreq " ] * 10 ) ) / int ( Config [ " InputConfig " ] [ " UseFreq " ] ) ) % 3600 ) % 60 )
else :
raise ValueError ( " 模式不存在 " )
self . ui . lineEdit_current_part_time . setText ( str ( hour ) + " 时 " + str ( min ) + " 分 " + str ( sec ) + " 秒 " )
self . ui . lineEdit_data_part_num . setText ( str ( Config [ " CurrentPartNum " ] ) + " / " + str ( Config [ " DataPartNum " ] ) )
def update_tableWidget ( self ) :
if Config [ " Mode " ] == " 30s " :
label_list = {
" label_a1 " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_30S_A1 ) [ 0 ] + 1 ,
" label_a2 " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_30S_A2 ) [ 0 ] + 1 ,
" label_b1 " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_30S_B1 ) [ 0 ] + 1 ,
" label_b2 " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_30S_B2 ) [ 0 ] + 1 ,
" label_c " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_30S_C ) [ 0 ] + 1 ,
" label_tobeLabeled " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_TOBELABELED ) [ 0 ] + 1
}
self . ui . tableWidget_a1 . setRowCount ( label_list [ " label_a1 " ] . __len__ ( ) )
self . ui . tableWidget_a2 . setRowCount ( label_list [ " label_a2 " ] . __len__ ( ) )
self . ui . tableWidget_b1 . setRowCount ( label_list [ " label_b1 " ] . __len__ ( ) )
self . ui . tableWidget_b2 . setRowCount ( label_list [ " label_b2 " ] . __len__ ( ) )
self . ui . tableWidget_c . setRowCount ( label_list [ " label_c " ] . __len__ ( ) )
self . ui . tableWidget_f . setRowCount ( label_list [ " label_tobeLabeled " ] . __len__ ( ) )
for label_name , label in label_list . items ( ) :
if label_name == " label_a1 " :
tableWidget = self . ui . tableWidget_a1
elif label_name == " label_a2 " :
tableWidget = self . ui . tableWidget_a2
elif label_name == " label_b1 " :
tableWidget = self . ui . tableWidget_b1
elif label_name == " label_b2 " :
tableWidget = self . ui . tableWidget_b2
elif label_name == " label_c " :
tableWidget = self . ui . tableWidget_c
elif label_name == " label_tobeLabeled " :
tableWidget = self . ui . tableWidget_f
else :
raise ValueError ( " 标签名不存在 " )
for row , value in enumerate ( label ) :
item = QTableWidgetItem ( str ( value ) . strip ( ) )
tableWidget . setItem ( row , 0 , item )
if ( self . data . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] [
value - 1 ] != Constants . STRING_IS_EMPTY and
self . data . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] [ value - 1 ] is not nan ) :
item = tableWidget . item ( row , 0 )
item . setBackground ( QColor ( 255 , 200 , 200 ) )
else :
item = tableWidget . item ( row , 0 )
item . setBackground ( QColor ( 255 , 255 , 255 ) )
self . ui . tableWidget_a1 . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_a1 . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_a2 . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_a2 . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_b1 . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_b2 . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_c . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_c . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_f . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_f . verticalScrollBar ( ) . maximum ( ) )
elif Config [ " Mode " ] == " 10s " :
label_list = {
" label_a " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_10S_A ) [ 0 ] + 1 ,
" label_b " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_10S_B ) [ 0 ] + 1 ,
" label_c " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_10S_C ) [ 0 ] + 1 ,
" label_tobeLabeled " : where ( self . data . label == Constants . BCG_QUALITY_LABEL_TOBELABELED ) [ 0 ] + 1
}
self . ui . tableWidget_a1 . setRowCount ( label_list [ " label_a " ] . __len__ ( ) )
self . ui . tableWidget_b1 . setRowCount ( label_list [ " label_b " ] . __len__ ( ) )
self . ui . tableWidget_c . setRowCount ( label_list [ " label_c " ] . __len__ ( ) )
self . ui . tableWidget_f . setRowCount ( label_list [ " label_tobeLabeled " ] . __len__ ( ) )
for label_name , label in label_list . items ( ) :
if label_name == " label_a " :
tableWidget = self . ui . tableWidget_a1
elif label_name == " label_b " :
tableWidget = self . ui . tableWidget_b1
elif label_name == " label_c " :
tableWidget = self . ui . tableWidget_c
elif label_name == " label_tobeLabeled " :
tableWidget = self . ui . tableWidget_f
else :
raise ValueError ( " 标签名不存在 " )
for row , value in enumerate ( label ) :
item = QTableWidgetItem ( str ( value ) . strip ( ) )
tableWidget . setItem ( row , 0 , item )
if ( self . data . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] [
value - 1 ] != Constants . STRING_IS_EMPTY and
self . data . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] [ value - 1 ] is not nan ) :
item = tableWidget . item ( row , 0 )
item . setBackground ( QColor ( 255 , 200 , 200 ) )
else :
item = tableWidget . item ( row , 0 )
item . setBackground ( QColor ( 255 , 255 , 255 ) )
self . ui . tableWidget_a1 . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_a1 . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_b1 . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_b1 . verticalScrollBar ( ) . maximum ( ) )
self . ui . tableWidget_c . verticalScrollBar ( ) . setValue ( self . ui . tableWidget_c . verticalScrollBar ( ) . maximum ( ) )
else :
raise ValueError ( " 模式不存在 " )
def reset_axes ( self ) :
if self . ax0 is not None :
self . ax0 . clear ( )
self . ax0 . grid ( True )
self . ax0 . xaxis . set_major_formatter ( ConfigParams . FORMATTER )
def change_tablewidget_mode ( self ) :
if Config [ " Mode " ] == " 10s " :
self . ui . pushButton_a1 . show ( )
self . ui . pushButton_a2 . hide ( )
self . ui . pushButton_b1 . show ( )
self . ui . pushButton_b2 . hide ( )
self . ui . pushButton_c . show ( )
self . ui . pushButton_f . show ( )
self . ui . pushButton_a1 . setText ( " a " )
self . ui . pushButton_a2 . setText ( " a2 " )
self . ui . pushButton_b1 . setText ( " b " )
self . ui . pushButton_b2 . setText ( " b2 " )
self . ui . pushButton_c . setText ( " c " )
self . ui . pushButton_f . setText ( " 删除 " )
self . ui . tableWidget_a1 . show ( )
self . ui . tableWidget_a2 . hide ( )
self . ui . tableWidget_b1 . show ( )
self . ui . tableWidget_b2 . hide ( )
self . ui . tableWidget_c . show ( )
self . ui . tableWidget_f . show ( )
self . ui . tableWidget_a1 . setHorizontalHeaderLabels ( [ ' a ' ] )
self . ui . tableWidget_a2 . setHorizontalHeaderLabels ( [ ' a2 ' ] )
self . ui . tableWidget_b1 . setHorizontalHeaderLabels ( [ ' b ' ] )
self . ui . tableWidget_b2 . setHorizontalHeaderLabels ( [ ' b2 ' ] )
self . ui . tableWidget_c . setHorizontalHeaderLabels ( [ ' c ' ] )
self . ui . tableWidget_f . setHorizontalHeaderLabels ( [ ' None ' ] )
elif Config [ " Mode " ] == " 30s " :
self . ui . pushButton_a1 . show ( )
self . ui . pushButton_a2 . show ( )
self . ui . pushButton_b1 . show ( )
self . ui . pushButton_b2 . show ( )
self . ui . pushButton_c . show ( )
self . ui . pushButton_f . show ( )
self . ui . pushButton_a1 . setText ( " a1 " )
self . ui . pushButton_a2 . setText ( " a2 " )
self . ui . pushButton_b1 . setText ( " b1 " )
self . ui . pushButton_b2 . setText ( " b2 " )
self . ui . pushButton_c . setText ( " c " )
self . ui . pushButton_f . setText ( " 删除 " )
self . ui . tableWidget_a1 . show ( )
self . ui . tableWidget_a2 . show ( )
self . ui . tableWidget_b1 . show ( )
self . ui . tableWidget_b2 . show ( )
self . ui . tableWidget_c . show ( )
self . ui . tableWidget_f . show ( )
self . ui . tableWidget_a1 . setHorizontalHeaderLabels ( [ ' a1 ' ] )
self . ui . tableWidget_a2 . setHorizontalHeaderLabels ( [ ' a2 ' ] )
self . ui . tableWidget_b1 . setHorizontalHeaderLabels ( [ ' b1 ' ] )
self . ui . tableWidget_b2 . setHorizontalHeaderLabels ( [ ' b2 ' ] )
self . ui . tableWidget_c . setHorizontalHeaderLabels ( [ ' c ' ] )
self . ui . tableWidget_f . setHorizontalHeaderLabels ( [ ' None ' ] )
elif Config [ " Mode " ] == " Undefined " :
return False
else :
raise ValueError ( " 模式不存在 " )
return True
class Data ( ) :
def __init__ ( self ) :
self . BCG = None
self . BCG_without_industrialFrequencyNoise = None
self . Artifact_a = None
self . df_label = None
self . label = None
self . artifact_number = array ( [ ] ) . astype ( int64 )
self . artifact_type = array ( [ ] ) . astype ( int64 )
self . artifact_mask = array ( [ ] ) . astype ( int64 )
def open_file ( self ) :
if Path ( Config [ " Path " ] [ " Input_BCG " ] ) . is_file ( ) :
Config [ " Path " ] [ " Input_BCG " ] = str ( Path ( Config [ " Path " ] [ " Input_BCG " ] ) . parent )
if Path ( Config [ " Path " ] [ " Input_Artifact " ] ) . is_file ( ) :
Config [ " Path " ] [ " Input_Artifact " ] = str ( Path ( Config [ " Path " ] [ " Input_Artifact " ] ) . parent )
if Path ( Config [ " Path " ] [ " Save " ] ) . is_file ( ) :
Config [ " Path " ] [ " Save " ] = str ( Path ( Config [ " Path " ] [ " Save " ] ) . parent )
result = PublicFunc . examine_file ( Config [ " Path " ] [ " Input_BCG " ] , ConfigParams . BCG_SYNC , ConfigParams . ENDSWITH_TXT )
if result . status :
Config [ " Path " ] [ " Input_BCG " ] = result . data [ " path " ]
Config [ " InputConfig " ] [ " ThoFreq " ] = result . data [ " freq " ]
else :
return result
result = PublicFunc . examine_file ( Config [ " Path " ] [ " Input_Artifact " ] , ConfigParams . ARTIFACT_A , ConfigParams . ENDSWITH_TXT )
if result . status :
Config [ " Path " ] [ " Input_Artifact " ] = result . data [ " path " ]
else :
return result
if Config [ " Mode " ] == " 30s " :
Config [ " Path " ] [ " Save " ] = str (
Path ( Config [ " Path " ] [ " Save " ] ) / Path ( ConfigParams . SQ_LABEL_30S + ConfigParams . ENDSWITH_CSV ) )
elif Config [ " Mode " ] == " 10s " :
Config [ " Path " ] [ " Save " ] = str (
Path ( Config [ " Path " ] [ " Save " ] ) / Path ( ConfigParams . SQ_LABEL_10S + ConfigParams . ENDSWITH_CSV ) )
elif Config [ " Mode " ] == " Undefined " :
return Result ( ) . failure ( info = Constants . INPUT_FAILURE + Constants . FAILURE_REASON [ " Mode_Undefined " ] )
else :
raise ValueError ( " 模式不存在 " )
try :
self . BCG = read_csv ( Config [ " Path " ] [ " Input_BCG " ] ,
encoding = ConfigParams . UTF8_ENCODING ,
header = None ) . to_numpy ( ) . reshape ( - 1 )
self . Artifact_a = read_csv ( Config [ " Path " ] [ " Input_Artifact " ] ,
encoding = ConfigParams . UTF8_ENCODING ,
header = None ) . to_numpy ( ) . reshape ( - 1 )
except Exception as e :
return Result ( ) . failure ( info = Constants . INPUT_FAILURE + Constants . FAILURE_REASON [ " Open_Data_Exception " ] + " \n " + format_exc ( ) )
try :
if Config [ " Mode " ] == " 30s " :
Config [ " DataPartNum " ] = len ( self . BCG ) / / ( Config [ " InputConfig " ] [ " UseFreq " ] * 30 )
elif Config [ " Mode " ] == " 10s " :
Config [ " DataPartNum " ] = len ( self . BCG ) / / ( Config [ " InputConfig " ] [ " UseFreq " ] * 10 )
else :
raise ValueError ( " 模式不存在 " )
if Config [ " DataPartNum " ] == 0 :
return Result ( ) . failure ( info = Constants . INPUT_FAILURE + Constants . FAILURE_REASON [ " Data_Length_Not_Correct " ] )
except Exception as e :
return Result ( ) . failure ( info = Constants . INPUT_FAILURE + Constants . FAILURE_REASON [ " Open_Data_Exception " ] + " \n " + format_exc ( ) )
try :
# 检查体动标签正确性,长度
PublicFunc . examine_artifact ( self . Artifact_a )
# 定义绘制体动所需要的数组
artifact_start = array ( [ ] )
artifact_start = artifact_start . astype ( int64 )
artifact_end = array ( [ ] )
artifact_end = artifact_end . astype ( int64 )
for i in range ( 0 , len ( self . Artifact_a ) , 4 ) :
self . artifact_number = append ( self . artifact_number , self . Artifact_a [ i ] )
for i in range ( 1 , len ( self . Artifact_a ) , 4 ) :
self . artifact_type = append ( self . artifact_type , self . Artifact_a [ i ] )
for i in range ( 2 , len ( self . Artifact_a ) , 4 ) :
artifact_start = append ( artifact_start , self . Artifact_a [ i ] )
for i in range ( 3 , len ( self . Artifact_a ) , 4 ) :
artifact_end = append ( artifact_end , self . Artifact_a [ i ] )
self . artifact_mask = zeros ( len ( self . BCG ) )
for i in range ( 0 , len ( self . artifact_number ) ) :
self . artifact_mask [ artifact_start [ i ] : artifact_end [ i ] + 1 ] = 1
except Exception as e :
return Result ( ) . failure ( info = Constants . INPUT_FAILURE +
Constants . FAILURE_REASON [ " Get_Artifact_Format_Exception " ] + " \n " + format_exc ( ) )
return Result ( ) . success ( info = Constants . INPUT_FINISHED )
def get_archive ( self ) :
if Config [ " Mode " ] == " 30s " :
filename = ConfigParams . SQ_LABEL_30S
elif Config [ " Mode " ] == " 10s " :
filename = ConfigParams . SQ_LABEL_10S
else :
raise ValueError ( " 模式不存在 " )
if not Path ( Config [ " Path " ] [ " Save " ] ) . exists ( ) :
self . label = full ( Config [ " DataPartNum " ] , Constants . BCG_QUALITY_LABEL_TOBELABELED )
self . df_label = DataFrame ( columns = [ Constants . BCG_QUALITY_LABEL_COLUMN_LABEL , Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] )
self . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_LABEL ] = self . label
self . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] = Constants . STRING_IS_EMPTY
self . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_LABEL ] = self . df_label [
Constants . BCG_QUALITY_LABEL_COLUMN_LABEL ] . astype ( str )
self . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] = self . df_label [
Constants . BCG_QUALITY_LABEL_COLUMN_REMARK ] . astype ( str )
return Result ( ) . success ( info = filename + " : " + Constants . ARCHIVE_NOT_EXIST )
else :
self . df_label = read_csv ( Config [ " Path " ] [ " Save " ] ,
encoding = ConfigParams . GBK_ENCODING )
self . label = self . df_label [ Constants . BCG_QUALITY_LABEL_COLUMN_LABEL ] . astype ( str )
return Result ( ) . success ( info = filename + " : " + Constants . ARCHIVE_EXIST )
def resample ( self ) :
if self . BCG is None :
Result ( ) . failure ( info = Constants . RESAMPLE_FAILURE + Constants . FAILURE_REASON [ " Data_Not_Exist " ] )
try :
if Config [ " InputConfig " ] [ " BCGFreq " ] != Config [ " InputConfig " ] [ " UseFreq " ] :
self . BCG = resample ( self . BCG ,
int ( len ( self . BCG ) *
( Config [ " InputConfig " ] [ " UseFreq " ] / Config [ " InputConfig " ] [ " BCGFreq " ] ) ) )
else :
return Result ( ) . success ( info = Constants . RESAMPLE_NO_NEED )
except Exception as e :
Result ( ) . failure ( info = Constants . RESAMPLE_FAILURE +
Constants . FAILURE_REASON [ " Resample_Exception " ] + " \n " + format_exc ( ) )
return Result ( ) . success ( info = Constants . RESAMPLE_FINISHED )
def preprocess ( self ) :
if self . BCG is None :
return Result ( ) . failure ( info = Constants . PREPROCESS_FAILURE + Constants . FAILURE_REASON [ " Data_Not_Exist " ] )
try :
self . BCG_without_industrialFrequencyNoise = Data . wipe_industrialFrequencyNoise ( self . BCG , int ( Config [ " InputConfig " ] [ " UseFreq " ] ) )
except Exception as e :
return Result ( ) . failure ( info = Constants . PREPROCESS_FAILURE +
Constants . FAILURE_REASON [ " Preprocess_Exception " ] + " \n " + format_exc ( ) )
return Result ( ) . success ( info = Constants . PREPROCESS_FINISHED )
@staticmethod
def wipe_industrialFrequencyNoise ( data , fs ) :
# 设计带阻滤波器( Notch Filter) 来滤除49-51Hz的噪声
nyq = 0.5 * fs # 奈奎斯特频率
low = 49 / nyq
high = 51 / nyq
b , a = iirfilter ( 4 , [ low , high ] , btype = ' bandstop ' , ftype = ' butter ' ) # 4阶巴特沃斯带阻滤波器
return lfilter ( b , a , data )