<s> from openmdao . lib . datatypes . api import Float , Dict , Array , List , Int , Bool <EOL> from openmdao . main . api import Component , Assembly <EOL> import numpy as np <EOL> import time <EOL> """<STR_LIT>""" <EOL> class PhaseController ( Component ) : <EOL> """<STR_LIT>""" <EOL> phase = Float ( iotype = "<STR_LIT>" ) <EOL> state = Bool ( iotype = "<STR_LIT>" ) <EOL> alpha = Float ( iotype = "<STR_LIT>" ) <EOL> beta = Float ( iotype = "<STR_LIT>" ) <EOL> def __init__ ( self , default_a , default_b , state = False ) : <EOL> super ( PhaseController , self ) . __init__ ( ) <EOL> self . state = state <EOL> self . default_a = default_a <EOL> self . default_b = default_b <EOL> def toggle ( self ) : <EOL> if self . state : <EOL> self . state = False <EOL> else : <EOL> self . state = True <EOL> return self . state <EOL> def on ( self ) : <EOL> if not self . state : <EOL> self . toggle ( ) <EOL> def off ( self ) : <EOL> if self . state : <EOL> self . toggle ( ) <EOL> def execute ( self ) : <EOL> if self . state : <EOL> t = ( np . sin ( self . phase ) + <NUM_LIT:1.> ) / <NUM_LIT> <EOL> t = <NUM_LIT> * t + <NUM_LIT:0.1> <EOL> self . alpha = t <EOL> self . beta = <NUM_LIT:1> - t <EOL> else : <EOL> self . alpha = self . default_a <EOL> self . beta = self . default_b <EOL> class BufferFFT ( Component ) : <EOL> """<STR_LIT>""" <EOL> ready = Bool ( False , iotype = "<STR_LIT>" ) <EOL> fps = Float ( iotype = "<STR_LIT>" ) <EOL> size = Int ( iotype = "<STR_LIT>" ) <EOL> n = Int ( iotype = "<STR_LIT>" ) <EOL> def __init__ ( self , n = <NUM_LIT> , spike_limit = <NUM_LIT> ) : <EOL> super ( BufferFFT , self ) . __init__ ( ) <EOL> self . n = n <EOL> self . add ( "<STR_LIT>" , Float ( iotype = "<STR_LIT>" ) ) <EOL> self . samples = [ ] <EOL> self . fps = <NUM_LIT:1.> <EOL> self . add ( "<STR_LIT>" , List ( iotype = "<STR_LIT>" ) ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . interpolated = np . zeros ( <NUM_LIT:2> ) <EOL> self . even_times = np . zeros ( <NUM_LIT:2> ) <EOL> self . spike_limit = spike_limit <EOL> def get_fft ( self ) : <EOL> n = len ( self . times ) <EOL> self . fps = float ( n ) / ( self . times [ - <NUM_LIT:1> ] - self . times [ <NUM_LIT:0> ] ) <EOL> self . even_times = np . linspace ( self . times [ <NUM_LIT:0> ] , self . times [ - <NUM_LIT:1> ] , n ) <EOL> interpolated = np . interp ( self . even_times , self . times , self . samples ) <EOL> interpolated = np . hamming ( n ) * interpolated <EOL> self . interpolated = interpolated <EOL> interpolated = interpolated - np . mean ( interpolated ) <EOL> fft = np . fft . rfft ( interpolated ) <EOL> self . freqs = float ( self . fps ) / n * np . arange ( n / <NUM_LIT:2> + <NUM_LIT:1> ) <EOL> return fft <EOL> def find_offset ( self ) : <EOL> N = len ( self . samples ) <EOL> for i in xrange ( <NUM_LIT:2> , N ) : <EOL> samples = self . samples [ i : ] <EOL> delta = max ( samples ) - min ( samples ) <EOL> if delta < self . spike_limit : <EOL> return N - i <EOL> def reset ( self ) : <EOL> N = self . find_offset ( ) <EOL> self . ready = False <EOL> self . times = self . times [ N : ] <EOL> self . samples = self . samples [ N : ] <EOL> def execute ( self ) : <EOL> self . samples . append ( self . data_in ) <EOL> self . times . append ( time . time ( ) ) <EOL> self . size = len ( self . samples ) <EOL> if self . size > self . n : <EOL> self . ready = True <EOL> self . samples = self . samples [ - self . n : ] <EOL> self . times = self . times [ - self . n : ] <EOL> if self . size > <NUM_LIT:4> : <EOL> self . fft = self . get_fft ( ) <EOL> if self . spike_limit : <EOL> if max ( self . samples ) - min ( self . samples ) > self . spike_limit : <EOL> self . reset ( ) <EOL> class bandProcess ( Component ) : <EOL> """<STR_LIT>""" <EOL> hz = Float ( iotype = "<STR_LIT>" ) <EOL> peak_hz = Float ( iotype = "<STR_LIT>" ) <EOL> phase = Float ( iotype = "<STR_LIT>" ) <EOL> def __init__ ( self , limits = [ <NUM_LIT:0.> , <NUM_LIT> ] , make_filtered = True , <EOL> operation = "<STR_LIT>" ) : <EOL> super ( bandProcess , self ) . __init__ ( ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . make_filtered = make_filtered <EOL> if make_filtered : <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . add ( "<STR_LIT>" , Array ( iotype = "<STR_LIT>" ) ) <EOL> self . limits = limits <EOL> self . operation = operation <EOL> def execute ( self ) : <EOL> if self . operation == "<STR_LIT>" : <EOL> idx = np . where ( ( self . freqs_in > self . limits [ <NUM_LIT:0> ] ) <EOL> & ( self . freqs_in < self . limits [ <NUM_LIT:1> ] ) ) <EOL> else : <EOL> idx = np . where ( ( self . freqs_in < self . limits [ <NUM_LIT:0> ] ) <EOL> & ( self . freqs_in > self . limits [ <NUM_LIT:1> ] ) ) <EOL> self . freqs = self . freqs_in [ idx ] <EOL> self . fft = np . abs ( self . fft_in [ idx ] ) ** <NUM_LIT:2> <EOL> if self . make_filtered : <EOL> fft_out = <NUM_LIT:0> * self . fft_in <EOL> fft_out [ idx ] = self . fft_in [ idx ] <EOL> if len ( fft_out ) > <NUM_LIT:2> : <EOL> self . filtered = np . fft . irfft ( fft_out ) <EOL> self . filtered = self . filtered / np . hamming ( len ( self . filtered ) ) <EOL> try : <EOL> maxidx = np . argmax ( self . fft ) <EOL> self . peak_hz = self . freqs [ maxidx ] <EOL> self . phase = np . angle ( self . fft_in ) [ idx ] [ maxidx ] <EOL> except ValueError : <EOL> pass <EOL> class Cardiac ( bandProcess ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , bpm_limits = [ <NUM_LIT:50> , <NUM_LIT> ] ) : <EOL> super ( Cardiac , self ) . __init__ ( ) <EOL> self . add ( "<STR_LIT>" , Float ( iotype = "<STR_LIT>" ) ) <EOL> self . limits = [ bpm_limits [ <NUM_LIT:0> ] / <NUM_LIT> , bpm_limits [ <NUM_LIT:1> ] / <NUM_LIT> ] <EOL> def execute ( self ) : <EOL> super ( Cardiac , self ) . execute ( ) <EOL> self . freqs = <NUM_LIT> * self . freqs <EOL> self . bpm = <NUM_LIT> * self . peak_hz </s>
