<s> import numpy as np <EOL> from . import common <EOL> class FIR ( object ) : <EOL> def __init__ ( self , h ) : <EOL> self . h = np . array ( h ) <EOL> self . x_state = [ <NUM_LIT:0> ] * len ( self . h ) <EOL> def __call__ ( self , x ) : <EOL> x_ = self . x_state <EOL> h = self . h <EOL> for v in x : <EOL> x_ = [ v ] + x_ [ : - <NUM_LIT:1> ] <EOL> yield np . dot ( x_ , h ) <EOL> self . x_state = x_ <EOL> class Demux ( object ) : <EOL> def __init__ ( self , sampler , omegas , Nsym ) : <EOL> self . Nsym = Nsym <EOL> self . filters = [ exp_iwt ( - w , Nsym ) / ( <NUM_LIT:0.5> * self . Nsym ) for w in omegas ] <EOL> self . filters = np . array ( self . filters ) <EOL> self . sampler = sampler <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> frame = self . sampler . take ( size = self . Nsym ) <EOL> if len ( frame ) == self . Nsym : <EOL> return np . dot ( self . filters , frame ) <EOL> else : <EOL> raise StopIteration <EOL> __next__ = next <EOL> def exp_iwt ( omega , n ) : <EOL> return np . exp ( <NUM_LIT> * omega * np . arange ( n ) ) <EOL> def norm ( x ) : <EOL> return np . sqrt ( np . dot ( x . conj ( ) , x ) . real ) <EOL> def rms ( x ) : <EOL> return np . mean ( np . abs ( x ) ** <NUM_LIT:2> , axis = <NUM_LIT:0> ) ** <NUM_LIT:0.5> <EOL> def coherence ( x , omega ) : <EOL> n = len ( x ) <EOL> Hc = exp_iwt ( - omega , n ) / np . sqrt ( <NUM_LIT:0.5> * n ) <EOL> norm_x = norm ( x ) <EOL> if norm_x : <EOL> return np . dot ( Hc , x ) / norm_x <EOL> else : <EOL> return <NUM_LIT:0.0> <EOL> def linear_regression ( x , y ) : <EOL> '''<STR_LIT>''' <EOL> x = np . array ( x ) <EOL> y = np . array ( y ) <EOL> mean_x = np . mean ( x ) <EOL> mean_y = np . mean ( y ) <EOL> x_ = x - mean_x <EOL> y_ = y - mean_y <EOL> a = np . dot ( y_ , x_ ) / np . dot ( x_ , x_ ) <EOL> b = mean_y - a * mean_x <EOL> return a , b <EOL> class MODEM ( object ) : <EOL> def __init__ ( self , symbols ) : <EOL> self . encode_map = { } <EOL> symbols = np . array ( list ( symbols ) ) <EOL> bits_per_symbol = np . log2 ( len ( symbols ) ) <EOL> bits_per_symbol = np . round ( bits_per_symbol ) <EOL> N = ( <NUM_LIT:2> ** bits_per_symbol ) <EOL> assert N == len ( symbols ) <EOL> bits_per_symbol = int ( bits_per_symbol ) <EOL> for i , v in enumerate ( symbols ) : <EOL> bits = [ int ( i & ( <NUM_LIT:1> << j ) != <NUM_LIT:0> ) for j in range ( bits_per_symbol ) ] <EOL> self . encode_map [ tuple ( bits ) ] = v <EOL> self . symbols = symbols <EOL> self . bits_per_symbol = bits_per_symbol <EOL> bits_map = dict ( item [ : : - <NUM_LIT:1> ] for item in self . encode_map . items ( ) ) <EOL> self . decode_list = [ ( s , bits_map [ s ] ) for s in self . symbols ] <EOL> def encode ( self , bits ) : <EOL> for bits_tuple in common . iterate ( bits , self . bits_per_symbol , tuple ) : <EOL> yield self . encode_map [ bits_tuple ] <EOL> def decode ( self , symbols , error_handler = None ) : <EOL> '''<STR_LIT>''' <EOL> symbols_vec = self . symbols <EOL> _dec = self . decode_list <EOL> for received in symbols : <EOL> error = np . abs ( symbols_vec - received ) <EOL> index = np . argmin ( error ) <EOL> decoded , bits = _dec [ index ] <EOL> if error_handler : <EOL> error_handler ( received = received , decoded = decoded ) <EOL> yield bits <EOL> def prbs ( reg , poly , bits ) : <EOL> '''<STR_LIT>''' <EOL> mask = ( <NUM_LIT:1> << bits ) - <NUM_LIT:1> <EOL> size = <NUM_LIT:0> <EOL> while ( poly >> size ) > <NUM_LIT:1> : <EOL> size += <NUM_LIT:1> <EOL> while True : <EOL> yield reg & mask <EOL> reg = reg << <NUM_LIT:1> <EOL> if reg >> size : <EOL> reg = reg ^ poly </s>
