<s> import core . modules <EOL> import core . modules . module_registry <EOL> from core . modules . vistrails_module import Module , ModuleError <EOL> from SciPy import SciPy <EOL> from Matrix import * <EOL> import scipy <EOL> from scipy import sparse , fftpack <EOL> import numpy <EOL> class DSP ( SciPy ) : <EOL> def compute ( self ) : <EOL> pass <EOL> class FFT ( DSP ) : <EOL> def compute ( self ) : <EOL> mat = self . get_input ( "<STR_LIT>" ) <EOL> pts = self . get_input ( "<STR_LIT>" ) <EOL> phasors = fftpack . fft ( mat . matrix . data , pts ) <EOL> outmat = sparse . csc_matrix ( phasors ) <EOL> out = SparseMatrix ( ) <EOL> out . matrix = outmat <EOL> self . set_output ( "<STR_LIT>" , out ) <EOL> class FFT2 ( DSP ) : <EOL> def compute ( self ) : <EOL> mat = self . get_input ( "<STR_LIT>" ) <EOL> phasors = fftpack . fftn ( mat . matrix . data ) <EOL> outmat = sparse . csc_matrix ( phasors ) <EOL> out = SparseMatrix ( ) <EOL> out . matrix = outmat <EOL> self . set_output ( "<STR_LIT>" , out ) <EOL> class WindowedFourierTransform ( DSP ) : <EOL> def compute ( self ) : <EOL> mat = self . get_input ( "<STR_LIT>" ) <EOL> sr = self . get_input ( "<STR_LIT>" ) <EOL> if self . has_input ( "<STR_LIT>" ) : <EOL> window = self . get_input ( "<STR_LIT>" ) <EOL> else : <EOL> window = sr <EOL> if self . has_input ( "<STR_LIT>" ) : <EOL> stride = self . get_input ( "<STR_LIT>" ) <EOL> else : <EOL> stride = int ( sr / <NUM_LIT:2> ) <EOL> signal_array = mat . matrix . toarray ( ) . ravel ( ) <EOL> pad = signal_array [ <NUM_LIT:0> : int ( window / <NUM_LIT:2> ) ] <EOL> signal_array = numpy . concatenate ( ( pad , signal_array ) ) <EOL> win_low = <NUM_LIT:0> <EOL> win_hi = window - <NUM_LIT:1> <EOL> phasors = fftpack . fft ( signal_array [ win_low : win_hi ] ) <EOL> out_array = phasors . ravel ( ) <EOL> win_low += stride <EOL> win_hi += stride <EOL> while win_hi < signal_array . shape [ <NUM_LIT:0> ] : <EOL> phasors = fftpack . fft ( signal_array [ win_low : win_hi ] ) <EOL> win_low += stride <EOL> win_hi += stride <EOL> out_array = numpy . vstack ( [ out_array , phasors . ravel ( ) ] ) <EOL> out = SparseMatrix ( ) <EOL> out . matrix = sparse . csc_matrix ( out_array ) <EOL> self . set_output ( "<STR_LIT>" , out ) <EOL> class ShortTimeFourierTransform ( DSP ) : <EOL> def get_signal ( self , sigs , window , offset , size ) : <EOL> win = scipy . zeros ( sigs . shape [ <NUM_LIT:0> ] ) . ravel ( ) <EOL> win [ offset : offset + size ] = window . ravel ( ) <EOL> part = sigs * win <EOL> return part <EOL> def compute ( self ) : <EOL> mat = self . get_input ( "<STR_LIT>" ) <EOL> sr = self . get_input ( "<STR_LIT>" ) <EOL> if self . has_input ( "<STR_LIT>" ) : <EOL> window = self . get_input ( "<STR_LIT>" ) . matrix . toarray ( ) <EOL> win_size = window . shape [ <NUM_LIT:1> ] <EOL> else : <EOL> win_size = self . get_input ( "<STR_LIT>" ) <EOL> window = scipy . signal . hamming ( win_size ) <EOL> if self . has_input ( "<STR_LIT>" ) : <EOL> stride = self . get_input ( "<STR_LIT>" ) <EOL> else : <EOL> stride = int ( win_size / <NUM_LIT:2> ) <EOL> signal_array = mat . matrix . transpose ( ) . toarray ( ) . ravel ( ) <EOL> samples = signal_array . shape [ <NUM_LIT:0> ] <EOL> offset = <NUM_LIT:0> <EOL> sig = self . get_signal ( signal_array , window , offset , win_size ) <EOL> phasors = fftpack . fft ( sig ) . ravel ( ) <EOL> out_array = phasors <EOL> offset += stride <EOL> i = <NUM_LIT:1> <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> sig = self . get_signal ( signal_array , window , offset , win_size ) <EOL> phasors = fftpack . fft ( sig ) <EOL> offset += stride <EOL> out_array = numpy . vstack ( [ out_array , phasors . ravel ( ) ] ) <EOL> i += <NUM_LIT:1> <EOL> except : <EOL> break <EOL> ( slices , freqs ) = out_array . shape <EOL> ar = out_array [ <NUM_LIT:0> : , <NUM_LIT:0> : sr * <NUM_LIT:2> ] <EOL> ar = ar [ <NUM_LIT:0> : , : : - <NUM_LIT:1> ] <EOL> out = SparseMatrix ( ) <EOL> sigout = SparseMatrix ( ) <EOL> sigout . matrix = sparse . csc_matrix ( signal_array ) <EOL> out . matrix = sparse . csc_matrix ( ar ) <EOL> self . set_output ( "<STR_LIT>" , sigout ) <EOL> self . set_output ( "<STR_LIT>" , out ) </s>
