${CMAKE_BINARY_DIR}/gnuradio-core/src/lib/swig

${CMAKE_BINARY_DIR}/gr-digital/python

${CMAKE_BINARY_DIR}/gr-digital/swig

)

set(GR_TEST_TARGET_DEPS volk gruel gnuradio-core gnuradio-digital)

GR_ADD_TEST(${py_qa_test_name} ${PYTHON_EXECUTABLE} ${PYTHON_DASH_B} ${py_qa_test_file})

__doc__ += shared_mod_args

def __init__(self, mod_code=None, *args, **kwargs):

# /////////////////////////////////////////////////////////////////////////////

# DBPSK demodulator

__doc__ += shared_demod_args

def __init__(self, mod_code=None, *args, **kwargs):

#

# Add these to the mod/demod registry

# See gnuradio-examples/python/digital for examples

from math import pi

import numpy

else:

raise TypeError, ("cpm_type must be an integer in {0,1,2,3}, is %r" % (cpm_type,))

# FM modulation

self.fmmod = gr.frequency_modulator_fc(sensitivity)

#

from gnuradio import gru

import struct

def gen_and_append_crc32(s):

return s + struct.pack(">I", gru.hexint(crc) & 0xFFFFFFFF)

def check_crc32(s):

return (False, '')

msg = s[:-4]

#print "msg = '%s'" % (msg,)

(expected,) = struct.unpack(">I", s[-4:])

# print "actual =", hex(actual), "expected =", hex(expected)

return (actual == expected, msg)

import digital_swig as digital

import math

# default values (used in __init__ and add_options)

_def_samples_per_symbol = 2

_def_excess_bw = 0.35

gr.io_signature(1, 1, gr.sizeof_char), # Input signature

gr.io_signature(1, 1, gr.sizeof_gr_complex)) # Output signature

self._samples_per_symbol = samples_per_symbol

self._excess_bw = excess_bw

self._differential = differential

1.0, # symbol rate

self._excess_bw, # excess bandwidth (roll-off factor)

ntaps)

# Connect

blocks = [self, self.bytes2chunks]

gr.io_signature(1, 1, gr.sizeof_gr_complex), # Input signature

gr.io_signature(1, 1, gr.sizeof_char)) # Output signature

self._samples_per_symbol = samples_per_symbol

self._excess_bw = excess_bw

self._phase_bw = phase_bw

fmin = -0.25

fmax = 0.25

self.receiver = digital.constellation_receiver_cb(

fmin, fmax)

# Do differential decoding based on phase change of symbols

from pprint import pprint

import inspect

# default values (used in __init__ and add_options)

_def_samples_per_symbol = 2

_def_sensitivity = 1

#sensitivity = (pi / 2) / samples_per_symbol # phase change per bit = pi / 2

# Turn it into NRZ data.

# Form Gaussian filter

# Generate Gaussian response (Needs to be convolved with window below).

self.sqwave = (1,) * samples_per_symbol # rectangular window

self.taps = numpy.convolve(numpy.array(self.gaussian_taps),numpy.array(self.sqwave))

# FM modulation

self.fmmod = gr.frequency_modulator_fc(sensitivity)

self._setup_logging()

# Connect & Initialize base class

def samples_per_symbol(self):

return self._samples_per_symbol

from pprint import pprint

import inspect

# default values (used in __init__ and add_options)

_def_samples_per_symbol = 2

_def_bt = 0.35

sensitivity = (pi / 2) / samples_per_symbol # phase change per bit = pi / 2

# Turn it into NRZ data.

# Form Gaussian filter

# Generate Gaussian response (Needs to be convolved with window below).

self.sqwave = (1,) * samples_per_symbol # rectangular window

self.taps = numpy.convolve(numpy.array(self.gaussian_taps),numpy.array(self.sqwave))

# FM modulation

self.fmmod = gr.frequency_modulator_fc(sensitivity)

self._setup_logging()

# Connect & Initialize base class

def samples_per_symbol(self):

return self._samples_per_symbol

#

import math

import ofdm_packet_utils

from ofdm_receiver import ofdm_receiver

import gnuradio.gr.gr_threading as _threading

constel = qam.qam_constellation(arity)

rotated_const = map(lambda pt: pt * rot, constel.points())

#print rotated_const

self.scale = gr.multiply_const_cc(1.0 / math.sqrt(self._fft_length))

self.connect((self._pkt_input, 0), (self.preambles, 0))

phgain = 0.25

frgain = phgain*phgain / 4.0

self.connect(self, self.ofdm_recv)

self.connect((self.ofdm_recv, 0), (self.ofdm_demod, 0))

from numpy import fft

from gnuradio import gr

from ofdm_sync_pn import ofdm_sync_pn

from ofdm_sync_fixed import ofdm_sync_fixed

from ofdm_sync_pnac import ofdm_sync_pnac

from ofdm_sync_ml import ofdm_sync_ml

class ofdm_receiver(gr.hier_block2):

"""

Performs receiver synchronization on OFDM symbols.

bw+tb, # midpoint of trans. band

tb, # width of trans. band

gr.firdes.WIN_HAMMING) # filter type

win = [1 for i in range(fft_length)]

self.nco = gr.frequency_modulator_fc(nco_sensitivity) # generate a signal proportional to frequency error of sync block

self.sigmix = gr.multiply_cc()

self.fft_demod = gr.fft_vcc(fft_length, True, win, True)

fft_length,

cp_length, ks[0])

import math

from gnuradio import gr

class ofdm_sync_ml(gr.hier_block2):

def __init__(self, fft_length, cp_length, snr, kstime, logging):

''' Maximum Likelihood OFDM synchronizer:

self.adder = gr.add_ff()

moving_sum_taps = [rho/2 for i in range(cp_length)]

self.connect(self.input,self.magsqrd1)

self.connect(self.delay,self.magsqrd2)

self.mixer = gr.multiply_cc();

movingsum2_taps = [1.0 for i in range(cp_length)]

# Correlator data handler

self.c2mag = gr.complex_to_mag()

# to readjust the timing in the middle of the packet or we ruin the equalizer settings.

kstime = [k.conjugate() for k in kstime]

kstime.reverse()

self.corrmag = gr.complex_to_mag_squared()

self.div = gr.divide_ff()

from numpy import fft

from gnuradio import gr

class ofdm_sync_pn(gr.hier_block2):

def __init__(self, fft_length, cp_length, logging=False):

"""

# Create a moving sum filter for the corr output

if 1:

moving_sum_taps = [1.0 for i in range(fft_length//2)]

else:

moving_sum_taps = [complex(1.0,0.0) for i in range(fft_length//2)]

# Create a moving sum filter for the input

self.inputmag2 = gr.complex_to_mag_squared()

movingsum2_taps = [1.0 for i in range(fft_length//2)]

if 1:

else:

self.square = gr.multiply_ff()

self.normalize = gr.divide_ff()

# Create a moving sum filter for the corr output

matched_filter_taps = [1.0/cp_length for i in range(cp_length)]

self.connect(self.normalize, self.matched_filter)

self.connect(self.matched_filter, self.sub1, self.pk_detect)

from numpy import fft

from gnuradio import gr

class ofdm_sync_pnac(gr.hier_block2):

def __init__(self, fft_length, cp_length, kstime, logging=False):

"""

# cross-correlate with the known symbol

kstime = [k.conjugate() for k in kstime[0:fft_length//2]]

kstime.reverse()

# Create a delay line

self.delay = gr.delay(gr.sizeof_gr_complex, fft_length/2)

# Create a moving sum filter for the input

self.mag = gr.complex_to_mag_squared()

movingsum_taps = (fft_length//1)*[1.0,]

# Get magnitude (peaks) and angle (phase/freq error)

self.c2mag = gr.complex_to_mag_squared()

from gnuradio import gr

import gnuradio.gr.gr_threading as _threading

import packet_utils

# /////////////////////////////////////////////////////////////////////////////

threshold = 12 # FIXME raise exception

self._rcvd_pktq = gr.msg_queue() # holds packets from the PHY

self.framer_sink = digital.framer_sink_1(self._rcvd_pktq)

self.connect(self, self._demodulator, self.correlator, self.framer_sink)

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

#

from gnuradio import gr, gr_unittest

import math, random

self.tb = None

expected_result = ( 0, 1, 0, 0, 1, 1, 0, 0, 0, 1, 1, 1, 0, 1, 1, 1, 1)

src_data = (-1, 1, -1, -1, 1, 1, -1, -1, -1, 1, 1, 1, -1, 1, 1, 1, 1)

src_data = [s + (1 - random.random()) for s in src_data] # add some noise

#print "actual result", actual_result

#print "expected result", expected_result

if __name__ == '__main__':

class test_chunks_to_symbols(gr_unittest.TestCase):

self.tb = None

def test_bc_001(self):

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

#

from gnuradio import gr, gr_unittest

import random, cmath

class test_clock_recovery_mm(gr_unittest.TestCase):

self.tb = None

# Test complex/complex version

omega = 2

gain_omega = 0.001

gain_mu = 0.01

omega_rel_lim = 0.001

data = 100*[complex(1, 1),]

self.src = gr.vector_source_c(data, False)

#print expected_result

#print dst_data

# Test float/float version

omega = 2

gain_omega = 0.01

gain_mu = 0.01

omega_rel_lim = 0.001

data = 100*[1,]

self.src = gr.vector_source_f(data, False)

#print expected_result

#print dst_data

# Test complex/complex version with varying input

omega = 2

gain_omega = 0.01

gain_mu = 0.1

omega_rel_lim = 0.0001

data = 1000*[complex(1, 1), complex(1, 1), complex(-1, -1), complex(-1, -1)]

self.src = gr.vector_source_c(data, False)

#print expected_result

#print dst_data

# Test float/float version

omega = 2

gain_omega = 0.01

gain_mu = 0.1

omega_rel_lim = 0.001

data = 1000*[1, 1, -1, -1]

self.src = gr.vector_source_f(data, False)

#print expected_result

#print dst_data

if __name__ == '__main__':

#

from gnuradio import gr, gr_unittest

class test_cma_equalizer_fir(gr_unittest.TestCase):

def transform(self, src_data):

SRC = gr.vector_source_c(src_data, False)

DST = gr.vector_sink_c()

self.tb.connect(SRC, EQU, DST)

self.tb.run()

src_data = (1+0j, 0+1j, -1+0j, 0-1j)*1000

expected_data = src_data

result = self.transform(src_data)

if __name__ == "__main__":

gr_unittest.run(test_cma_equalizer_fir, "test_cma_equalizer_fir.xml")

from gnuradio import gr, gr_unittest, blks2

from utils import mod_codes

# import from local folder

import psk

(-1+0j), (0-1j))

rot_sym = 2

dim = 2

def threed_constell():

oned_points = ((1+0j), (0+1j), (-1+0j), (0-1j))

points += [oned_points[ia], oned_points[ib], oned_points[ic]]

rot_sym = 4

dim = 3

tested_constellation_info = (

(psk.psk_constellation,

'mod_code': tested_mod_codes,

'differential': (False,)},

False, None),

(twod_constell, {}, True, None),

(threed_constell, {}, True, None),

)

break

src_length = 256

data = dst.data()

# Don't worry about cut off data for now.

first = constellation.bits_per_symbol()

class mod_demod(gr.hier_block2):

self.blocks = [self]

# We expect a stream of unpacked bits.

# First step is to pack them.

# Second step we unpack them such that we have k bits in each byte where

# each constellation symbol hold k bits.

self.blocks.append(

# Apply any pre-differential coding

# Gray-coding is done here if we're also using differential coding.

if self.constellation.apply_pre_diff_code():

# Differential encoding.

if self.differential:

# Convert to constellation symbols.

# CHANNEL

# Channel just consists of a rotation to check differential coding.

if rotation is not None:

# RX

# Convert the constellation symbols back to binary values.

# Differential decoding.

if self.differential:

# Decode any pre-differential coding.

if self.constellation.apply_pre_diff_code():

mod_codes.invert_code(self.constellation.pre_diff_code())))

# unpack the k bit vector into a stream of bits

self.blocks.append(gr.unpack_k_bits_bb(

self.blocks.append(self)

self.connect(*self.blocks)

if __name__ == '__main__':

gr_unittest.run(test_constellation, "test_constellation.xml")

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

#

from gnuradio import gr, gr_unittest

import math

self.tb = None

src_data = (0.5 + 0.5j, 0.1 - 1.2j, -0.8 - 0.1j, -0.45 + 0.8j,

0.8 + 1.0j, -0.5 + 0.1j, 0.1 - 1.2j)

expected_result = ( 1, 1, 0, 0,

1, 0, 1)

#print "actual result", actual_result

#print "expected result", expected_result

src_data = (0.5 + 0.5j, 0.1 - 1.2j, -0.8 - 0.1j, -0.45 + 0.8j,

0.8 + 1.0j, -0.5 + 0.1j, 0.1 - 1.2j)

expected_result = ( 3, 1, 0, 2,

3, 2, 1)

#print "actual result", actual_result

#print "expected result", expected_result

if __name__ == '__main__':

from gnuradio import gr, blks2, gr_unittest

from utils import mod_codes, alignment

from generic_mod_demod import generic_mod, generic_demod

from qa_constellation import tested_constellations, twod_constell

PHASE_BW = 2*math.pi/100.0

# We ignore the first half of the output data since often it takes

# a while for the receiver to lock on.

self.assertTrue(correct > REQ_CORRECT)

"""

Takes a constellation an runs a generic modulation, channel,

and generic demodulation.

mod = generic_mod(constellation, differential=differential)

# Channel

if freq_offset:

else:

# Receiver Blocks

if freq_offset:

demod = generic_demod(constellation, differential=differential,

# 0 0 0 1 0 0 0 1

src_data = (1, 0, 1, 1, 1, 1, 0, 1, 1) + pad + (0,) * 7

expected_result = pad + (1, 0, 1, 1, 3, 1, 0, 1, 1, 2) + (0,) * 6

op = digital.correlate_access_code_bb("1011", 0)

def test_002(self):

#print access_code

src_data = code + (1, 0, 1, 1) + pad

expected_result = pad + code + (3, 0, 1, 1)

op = digital.correlate_access_code_bb(access_code, 0)

def test_003(self):

code = tuple(string_to_1_0_list(default_access_code))

#print access_code

src_data = code + (1, 0, 1, 1) + pad

expected_result = code + (1, 0, 1, 1) + pad

op = digital.correlate_access_code_tag_bb(access_code, 0, "test")

if __name__ == '__main__':

gr_unittest.run(test_correlate_access_code, "test_correlate_access_code.xml")

#

from gnuradio import gr, gr_unittest

import random, cmath

class test_costas_loop_cc(gr_unittest.TestCase):

self.tb = None

# test basic functionality by setting all gains to 0

natfreq = 0.0

order = 2

data = 100*[complex(1,0),]

self.src = gr.vector_source_c(data, False)

expected_result = data

dst_data = self.snk.data()

# Make sure it doesn't diverge given perfect data

natfreq = 0.25

order = 2

data = [complex(2*random.randint(0,1)-1, 0) for i in xrange(100)]

self.src = gr.vector_source_c(data, False)

expected_result = data

dst_data = self.snk.data()

# BPSK Convergence test with static rotation

natfreq = 0.25

order = 2

rot = cmath.exp(0.2j) # some small rotation

data = [complex(2*random.randint(0,1)-1, 0) for i in xrange(100)]

# generously compare results; the loop will converge near to, but

# not exactly on, the target data

# QPSK Convergence test with static rotation

natfreq = 0.25

order = 4

rot = cmath.exp(0.2j) # some small rotation

data = [complex(2*random.randint(0,1)-1, 2*random.randint(0,1)-1)

# generously compare results; the loop will converge near to, but

# not exactly on, the target data

# 8PSK Convergence test with static rotation

natfreq = 0.25

order = 8

rot = cmath.exp(-cmath.pi/8.0j) # rotate to match Costas rotation

const = psk.psk_constellation(order)

# generously compare results; the loop will converge near to, but

# not exactly on, the target data

if __name__ == '__main__':

gr_unittest.run(test_costas_loop_cc, "test_costas_loop_cc.xml")

#

from gnuradio import gr, gr_unittest

import numpy

class test_cpm(gr_unittest.TestCase):

self.tb = None

def do_check_phase_shift(self, type, name):

L = 1

in_bits = (1,) * 20

src = gr.vector_source_b(in_bits, False)

arg = gr.complex_to_arg()

sink = gr.vector_sink_f()

bt = 0.3

in_bits = (1,) * 20

src = gr.vector_source_b(in_bits, False)

arg = gr.complex_to_arg()

sink = gr.vector_sink_f()

#

from gnuradio import gr, gr_unittest

import random, cmath

class test_crc32(gr_unittest.TestCase):

self.tb = None

data = 100*"0"

expected_result = 2943744955

#print hex(result)

data = 100*"1"

expected_result = 2326594156

#print hex(result)

data = 10*"0123456789"

expected_result = 3774345973

#print hex(result)

if __name__ == '__main__':

gr_unittest.run(test_crc32, "test_crc32.xml")

return tuple(result)

self.tb = None

def test_diff_encdec_000(self):

import digital_swig as digital

import math

self.tb = None

src_data = (0+0j, 1+0j, -1+0j, 3+4j, -3-4j, -3+4j)

expected_result = (0+0j, 0+0j, -1+0j, -3-4j, -25+0j, -7-24j)

if __name__ == '__main__':

gr_unittest.run(test_diff_phasor, "test_diff_phasor.xml")

#

from gnuradio import gr, gr_unittest

class test_digital(gr_unittest.TestCase):

self.tb = None

if __name__ == '__main__':

#

from gnuradio import gr, gr_unittest

import random, math

class test_fll_band_edge_cc(gr_unittest.TestCase):

self.tb = None

sps = 4

rolloff = 0.35

bw = 2*math.pi/100.0

random.seed(0)

data = [2.0*random.randint(0, 2) - 1.0 for i in xrange(200)]

self.src = gr.vector_source_c(data, False)

# Mix symbols with a complex sinusoid to spin them

self.nco = gr.sig_source_c(1, gr.GR_SIN_WAVE, foffset, 1)

self.mix = gr.multiply_cc()

# FLL will despin the symbols to an arbitrary phase

# Create sinks for all outputs of the FLL

# we will only care about the freq and error outputs

dst_data = self.vsnk_frq.data()[N:]

expected_result = len(dst_data)* [-0.20,]

if __name__ == '__main__':

gr_unittest.run(test_fll_band_edge_cc, "test_fll_band_edge_cc.xml")

self.tb.connect(src, correlator, framer_sink)

self.tb.connect(correlator, vsnk)

result_data = rcvd_pktq.delete_head()

result_data = result_data.to_string()

def test_002(self):

self.tb.connect(src, correlator, framer_sink)

self.tb.connect(correlator, vsnk)

result_data = rcvd_pktq.delete_head()

result_data = result_data.to_string()

if __name__ == '__main__':

gr_unittest.run(test_framker_sink, "test_framker_sink.xml")

class test_glfsr_source(gr_unittest.TestCase):

self.tb = None

def test_000_make_b(self):

#

from gnuradio import gr, gr_unittest

class test_lms_dd_equalizer(gr_unittest.TestCase):

def transform(self, src_data, gain, const):

SRC = gr.vector_source_c(src_data, False)

DST = gr.vector_sink_c()

self.tb.connect(SRC, EQU, DST)

self.tb.run()

def test_001_identity(self):

# Constant modulus signal so no adjustments

src_data = const.points()*1000

N = 100 # settling time

expected_data = src_data[N:]

result = self.transform(src_data, 0.1, const)[N:]

if __name__ == "__main__":

gr_unittest.run(test_lms_dd_equalizer, "test_lms_dd_equalizer.xml")

def helper(self, symbols):

src_data = [0, 1, 2, 3, 0, 1, 2, 3]

expected_data = map(lambda x: symbols[x], src_data)

op = digital.map_bb(symbols)

result_data = list(dst.data())

def test_001(self):

symbols = [0, 0, 0, 0]

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

#

from gnuradio import gr, gr_unittest

class test_mpsk_receiver(gr_unittest.TestCase):

self.tb = None

# Test BPSK sync

M = 2

theta = 0

loop_bw = cmath.pi/100.0

fmin = -0.5

fmax = 0.5

gain_mu = 0.01

omega = 2

gain_omega = 0.001

omega_rel = 0.001

self.src = gr.vector_source_c(data, False)

self.snk = gr.vector_sink_c()

self.tb.run()

dst_data = self.snk.data()

# Only compare last Ncmp samples

len_e = len(expected_result)

len_d = len(dst_data)

dst_data = dst_data[len_d - Ncmp:]

#for e,d in zip(expected_result, dst_data):

# Test QPSK sync

M = 4

theta = 0

fmin = -0.5

fmax = 0.5

gain_mu = 0.01

omega = 2

gain_omega = 0.001

omega_rel = 0.001

self.src = gr.vector_source_c(data, False)

self.snk = gr.vector_sink_c()

self.tb.run()

# Only compare last Ncmp samples

len_e = len(expected_result)

len_d = len(dst_data)

dst_data = dst_data[len_d - Ncmp:]

#for e,d in zip(expected_result, dst_data):

if __name__ == '__main__':

gr_unittest.run(test_mpsk_receiver, "test_mpsk_receiver.xml")

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

def get_n_cplx():

return complex(random.random()-0.5, random.random()-0.5)

random.seed(0) # make repeatable

N = 10000

self._noise = [get_n_cplx() for i in xrange(N)]

self._bits = [get_cplx() for i in xrange(N)]

self.tb = None

result = []

for i in xrange(1,6):

src_data = [b+(i*n) for b,n in zip(self._bits, self._noise)]

result.append(op.snr())

return result

expected_result = [11.48, 5.91, 3.30, 2.08, 1.46]

N = 10000

alpha = 0.001

actual_result = self.mpsk_snr_est_setup(op)

expected_result = [11.48, 5.91, 3.30, 2.08, 1.46]

N = 10000

alpha = 0.001

actual_result = self.mpsk_snr_est_setup(op)

expected_result = [11.02, 6.20, 4.98, 5.16, 5.66]

N = 10000

alpha = 0.001

actual_result = self.mpsk_snr_est_setup(op)

expected_result = [10.90, 6.00, 4.76, 4.97, 5.49]

N = 10000

alpha = 0.001

actual_result = self.mpsk_snr_est_setup(op)

expected_result = [11.02, 6.20, 4.98, 5.16, 5.66]

actual_result = []

for i in xrange(1,6):

src_data = [b+(i*n) for b,n in zip(self._bits, self._noise)]

N = 10000

alpha = 0.001

actual_result.append(op.snr())

if __name__ == '__main__':

# Test various SNR estimators; we're not using a Gaussian

#!/usr/bin/env python

#

#

# This file is part of GNU Radio

#

from gnuradio import gr, gr_unittest

from pprint import pprint

self.tb = None

def helper(self, v0, v1, fft_length, preamble):

# print "len(v) = %d" % (len(v))

v2s = gr.vector_to_stream(gr.sizeof_gr_complex, fft_length)

dst0 = gr.vector_sink_c()

p.append(tuple(t))

v += t

r = self.helper(v, npayloads*[1], fft_length, preamble)

self.assertEqual(r[0], tuple(npayloads*[1, 0]))

p0, p1, p[12], p[13],

p0, p1, p[14], p[15]))

if __name__ == '__main__':

gr_unittest.run(test_ofdm_insert_preamble, "test_ofdm_insert_preamble.xml")

#

from gnuradio import gr, gr_unittest

import digital_swig as digital

import random, cmath

class test_pfb_clock_sync(gr_unittest.TestCase):

self.tb = None

# Test BPSK sync

excess_bw = 0.35

max_rate_deviation,

osps)

self.src = gr.vector_source_c(data, False)

# pulse shaping interpolation filter

1.0, # symbol rate

excess_bw, # excess bandwidth (roll-off factor)

ntaps)

self.snk = gr.vector_sink_c()

self.tb.connect(self.src, self.rrc_filter, self.test, self.snk)

self.tb.run()

dst_data = self.snk.data()

# Only compare last Ncmp samples

len_e = len(expected_result)

len_d = len(dst_data)

expected_result = expected_result[len_e - Ncmp:]

#for e,d in zip(expected_result, dst_data):

# print e, d

# Test real BPSK sync

excess_bw = 0.35

max_rate_deviation,

osps)

self.src = gr.vector_source_f(data, False)

# pulse shaping interpolation filter

1.0, # symbol rate

excess_bw, # excess bandwidth (roll-off factor)

ntaps)

self.snk = gr.vector_sink_f()

self.tb.connect(self.src, self.rrc_filter, self.test, self.snk)

self.tb.run()

dst_data = self.snk.data()

# Only compare last Ncmp samples

len_e = len(expected_result)

len_d = len(dst_data)

expected_result = expected_result[len_e - Ncmp:]

#for e,d in zip(expected_result, dst_data):

# print e, d

if __name__ == '__main__':

class test_pn_correlator_cc(gr_unittest.TestCase):

def setUp(self):

def tearDown(self):

self.tb = None

def test_001(self):

src_data = [0, 1, 0, 1]

expected_data = 1

op = digital.probe_density_b(1)

result_data = op.density()

def test_002(self):

src_data = [1, 1, 1, 1]

expected_data = 1

op = digital.probe_density_b(0.01)

result_data = op.density()

def test_003(self):

src_data = [0, 1, 0, 1, 0, 1, 0, 1, 0, 1]

expected_data = 0.95243

op = digital.probe_density_b(0.01)

result_data = op.density()

print result_data

if __name__ == '__main__':

gr_unittest.run(test_probe_density, "test_probe_density.xml")

class test_scrambler(gr_unittest.TestCase):

self.tb = gr.top_block()

def tearDown(self):

import digital_swig as digital

import math

self.tb = None

src_data = (0x00, 0x11, 0x22, 0x33,

0x44, 0x55, 0x66, 0x77,

0x88, 0x99, 0xaa, 0xbb,

0xac, 0xdd, 0xa4, 0xe2, 0xf2, 0x8c, 0x20, 0xfc, 0x02, 0x88, 0x99, 0xaa, 0xbb, 0x55,

0xac, 0xdd, 0xa4, 0xe2, 0xf2, 0x8c, 0x20, 0xfc, 0x03, 0xcc, 0xdd, 0xee, 0xff, 0x55)

if __name__ == '__main__':

gr_unittest.run(test_simple_framer, "test_simple_framer.xml")

from utils.gray_code import gray_code

from utils import mod_codes

import modulation_utils

# Default number of points in constellation.

_def_constellation_points = 16

pre_diff_code = range(0, m/2) + range(3*m/4, m) + range(m/2, 3*m/4)

else:

pre_diff_code = []

return constellation

# /////////////////////////////////////////////////////////////////////////////

from gnuradio.digital.generic_mod_demod import generic_mod, generic_demod

from gnuradio.digital.generic_mod_demod import shared_mod_args, shared_demod_args

from utils import mod_codes

import modulation_utils

# The default encoding (e.g. gray-code, set-partition)

"""

if mod_code != mod_codes.GRAY_CODE:

raise ValueError("This QPSK mod/demod works only for gray-coded constellations.")

# /////////////////////////////////////////////////////////////////////////////

# QPSK modulator

def __init__(self, mod_code=_def_mod_code, differential=False, *args, **kwargs):

pre_diff_code = True

if not differential:

if mod_code != mod_codes.GRAY_CODE:

raise ValueError("This QPSK mod/demod works only for gray-coded constellations.")

else:

raise ValueError("That mod_code is not supported for DQPSK mod/demod.")

if mod_code == mod_codes.NO_CODE:

pre_diff_code = False

*args, **kwargs):

pre_diff_code = True

if not differential:

if mod_code != mod_codes.GRAY_CODE:

raise ValueError("This QPSK mod/demod works only for gray-coded constellations.")

else:

raise ValueError("That mod_code is not supported for DQPSK mod/demod.")

if mod_code == mod_codes.NO_CODE:

pre_diff_code = False

def dqpsk_constellation(mod_code=_def_mod_code):

if mod_code != mod_codes.GRAY_CODE:

raise ValueError("The DQPSK constellation is only generated for gray_coding. But it can be used for non-grayed coded modulation if one doesn't use the pre-differential code.")

# /////////////////////////////////////////////////////////////////////////////

# DQPSK modulator

__doc__ += shared_mod_args

def __init__(self, mod_code=_def_mod_code, *args, **kwargs):

*args, **kwargs)

# /////////////////////////////////////////////////////////////////////////////

__doc__ += shared_demod_args

def __init__(self, mod_code=_def_mod_code, *args, **kwargs):

*args, **kwargs)

#

modulation_utils.add_type_1_mod('dqpsk', dqpsk_mod)

modulation_utils.add_type_1_demod('dqpsk', dqpsk_demod)

modulation_utils.add_type_1_constellation('dqpsk', dqpsk_constellation)