<s> __author__ = '<STR_LIT>' <EOL> import scipy . stats as stats <EOL> from PeakAnalysis import * <EOL> from . . utils . UMath import * <EOL> from . . Path import Path <EOL> class FeatureExtractor : <EOL> def __init__ ( self , output_path , view , use_statistical_features = False ) : <EOL> self . output_path = output_path <EOL> self . view = view <EOL> self . use_statistical_features = use_statistical_features <EOL> def segment_heuristically ( self , sensors , reference_signal , label = None ) : <EOL> p = PeakAnalysis ( self . view ) <EOL> peaks = p . get_peaks ( reference_signal ) <EOL> timestamps = [ ] <EOL> for peak_position in peaks : <EOL> timestamps . append ( sensors [ <NUM_LIT:0> ] . timestamp [ peak_position ] ) <EOL> if label : <EOL> label_timestamps = label . timestamp <EOL> labels = label . label <EOL> closer_labels = [ ] <EOL> closer_timestamps = [ ] <EOL> for i in range ( <NUM_LIT:0> , len ( label_timestamps ) ) : <EOL> min_dist = label_timestamps [ len ( label_timestamps ) - <NUM_LIT:1> ] - label_timestamps [ <NUM_LIT:0> ] <EOL> closer_label = None <EOL> closer_timestamp = None <EOL> for heuristic_timestamp in timestamps : <EOL> dist = sqrt ( np . square ( label_timestamps [ i ] - heuristic_timestamp ) ) <EOL> if dist < min_dist : <EOL> min_dist = dist <EOL> closer_label = labels [ i ] <EOL> closer_timestamp = heuristic_timestamp <EOL> closer_labels . append ( closer_label ) <EOL> closer_timestamps . append ( closer_timestamp ) <EOL> self . segment ( sensors , closer_timestamps , closer_labels ) <EOL> else : <EOL> self . segment ( sensors , timestamps ) <EOL> def segment_from_labels ( self , sensors , label ) : <EOL> label_timestamps = label . timestamp <EOL> labels = label . label <EOL> self . segment ( sensors , label_timestamps , labels ) <EOL> def segment ( self , sensors , timestamps , labels = None ) : <EOL> output_file = open ( "<STR_LIT>" . format ( self . output_path ) , '<STR_LIT:w>' ) <EOL> for i in range ( <NUM_LIT:0> , len ( timestamps ) ) : <EOL> features = self . get_features ( sensors , timestamps [ i ] ) <EOL> if labels is not None : <EOL> output_file . write ( "<STR_LIT>" . format ( labels [ i ] ) ) <EOL> else : <EOL> output_file . write ( "<STR_LIT>" ) <EOL> if not self . use_statistical_features : <EOL> for j in range ( <NUM_LIT:0> , len ( features [ <NUM_LIT:0> ] ) ) : <EOL> line = self . get_line ( features , j ) <EOL> output_file . write ( line ) <EOL> else : <EOL> features = np . array ( features ) . flatten ( ) <EOL> line = self . get_line ( features ) <EOL> output_file . write ( line ) <EOL> output_file . write ( "<STR_LIT:\n>" ) <EOL> output_file . close ( ) <EOL> print "<STR_LIT>" . format ( output_file . name ) <EOL> for sensor in sensors : <EOL> self . plot_segmentation ( sensor , timestamps , labels ) <EOL> def get_line ( self , features , j = None , separator = "<STR_LIT:U+002C>" ) : <EOL> length = len ( features ) <EOL> line = "<STR_LIT>" <EOL> for k in range ( <NUM_LIT:0> , length ) : <EOL> value = '<STR_LIT>' . format ( features [ k ] if j is None else features [ k ] [ j ] ) <EOL> line += "<STR_LIT>" . format ( value , separator if k < length - <NUM_LIT:1> else '<STR_LIT:\n>' ) <EOL> return line <EOL> def plot_segmentation ( self , sensor , label_timestamps , labels = None ) : <EOL> title = "<STR_LIT>" . format ( sensor . name ) <EOL> self . view . plot_sensor_data_and_label ( title . title ( ) , sensor . timestamp , sensor . x , sensor . y , sensor . z , label_timestamps , labels ) <EOL> self . view . save ( "<STR_LIT>" . format ( Path . RESULT_PATH , sensor . id , title . replace ( "<STR_LIT:U+0020>" , "<STR_LIT:_>" ) ) ) <EOL> self . view . show ( ) <EOL> def get_features ( self , sensors , timestamp_reference ) : <EOL> sensor = sensors [ <NUM_LIT:0> ] <EOL> center_timestamp_index = ( np . abs ( sensor . timestamp - timestamp_reference ) ) . argmin ( ) <EOL> features = [ ] <EOL> for i in range ( <NUM_LIT:0> , len ( sensors ) ) : <EOL> sensor = sensors [ i ] <EOL> sensor . normalize ( ) <EOL> if sensor . mean_signal is None : <EOL> x_sample = self . get_data_slice ( sensor . x , center_timestamp_index ) <EOL> y_sample = self . get_data_slice ( sensor . y , center_timestamp_index ) <EOL> z_sample = self . get_data_slice ( sensor . z , center_timestamp_index ) <EOL> features . append ( x_sample if not self . use_statistical_features else self . get_statistical_features ( x_sample ) ) <EOL> features . append ( y_sample if not self . use_statistical_features else self . get_statistical_features ( y_sample ) ) <EOL> features . append ( z_sample if not self . use_statistical_features else self . get_statistical_features ( z_sample ) ) <EOL> else : <EOL> mean_sample = self . get_data_slice ( sensor . mean_signal , center_timestamp_index ) <EOL> features . append ( mean_sample if not self . use_statistical_features else self . get_statistical_features ( mean_sample ) ) <EOL> return features <EOL> def get_statistical_features ( self , data ) : <EOL> p = PeakAnalysis ( self . view ) <EOL> min_value = np . amin ( data ) <EOL> max_value = np . amax ( data ) <EOL> root_mean_square = UMath . get_root_mean_square ( data ) <EOL> peaks_number = np . mean ( len ( p . get_peaks ( data ) ) ) <EOL> crest_factor = np . median ( p . get_peak_to_average_ratios ( data ) ) <EOL> skewness = stats . skew ( data , False ) <EOL> kurtosis = stats . kurtosis ( data , False ) <EOL> variance = np . var ( data ) <EOL> return [ min_value , max_value , root_mean_square , peaks_number , crest_factor , skewness , kurtosis , variance ] <EOL> def get_data_slice ( self , data , center_index , window_size = <NUM_LIT:50> ) : <EOL> left_samples = data [ center_index - ( window_size / <NUM_LIT:2> ) : center_index ] <EOL> right_samples = data [ center_index : center_index + ( window_size / <NUM_LIT:2> ) ] <EOL> return np . hstack ( ( left_samples , right_samples ) ) </s>
