"""

import math

import numpy

from gnuradio import gr, digital

N_BITS = 1e7

RAND_SEED = 42

def berawgn(EbN0):

""" Calculates theoretical bit error rate in AWGN (for BPSK and given Eb/N0) """

return 0.5 * erfc(math.sqrt(10**(float(EbN0) / 10)))

class BitErrors(gr.hier_block2):

""" Two inputs: true and received bits. We compare them and

add up the number of incorrect bits. Because integrate_ff()

can only add up a certain number of values, the output is

not a scalar, but a sequence of values, the sum of which is

the BER. """

def __init__(self, bits_per_byte):

gr.hier_block2.__init__(self, "BitErrors",

# Bit comparison

comp = blocks.xor_bb()

self)

self.connect((self, 1), (comp, 1))

class BERAWGNSimu(gr.top_block):

" This contains the simulation flow graph "

def __init__(self, EbN0):

gr.top_block.__init__(self)

self.const = digital.qpsk_constellation()

# Source is N_BITS bits, non-repeated

noise = analog.noise_source_c(analog.GR_GAUSSIAN,

self.EbN0_to_noise_voltage(EbN0),

RAND_SEED)

demod = digital.constellation_decoder_cb(self.const.base())

self.connect(src, mod, add, demod, ber, self.sink)

self.connect(noise, (add, 1))

self.connect(src, (ber, 1))

def EbN0_to_noise_voltage(self, EbN0):

""" Converts Eb/N0 to a complex noise voltage (assuming unit symbol power) """

def simulate_ber(EbN0):

fg.run()

return numpy.sum(fg.sink.data())

if __name__ == "__main__":

EbN0_min = 0

EbN0_max = 15

print("Simulating...")

f = pyplot.figure()

s.semilogy(EbN0_range, ber_theory, 'g-.', label="Theoretical")

s.semilogy(EbN0_range, ber_simu, 'b-o', label="Simulated")

s.set_title('BER Simulation')

print("Error: could not import pyplot (http://matplotlib.sourceforge.net/)")

sys.exit(1)

class example_costas(gr.top_block):

def __init__(self, N, sps, rolloff, ntaps, bw, noise, foffset, toffset, poffset):

gr.top_block.__init__(self)

rrc_taps = filter.firdes.root_raised_cosine(

sps, sps, 1.0, rolloff, ntaps)

self.src = blocks.vector_source_c(data.tolist(), False)

self.rrc = filter.interp_fir_filter_ccf(sps, rrc_taps)

self.connect(self.src, self.rrc, self.chn, self.cst, self.vsnk_cst)

self.connect(self.rrc, self.vsnk_src)

def main():

parser = ArgumentParser(conflict_handler="resolve")

parser.add_argument("-N", "--nsamples", type=int, default=2000,

parser.add_argument("-S", "--sps", type=int, default=4,

parser.add_argument("-r", "--rolloff", type=eng_float, default=0.35,

parser.add_argument("-n", "--ntaps", type=int, default=45,

parser.add_argument("--noise", type=eng_float, default=0.0,

parser.add_argument("-f", "--foffset", type=eng_float, default=0.0,

parser.add_argument("-t", "--toffset", type=eng_float, default=1.0,

parser.add_argument("-p", "--poffset", type=eng_float, default=0.707,

args = parser.parse_args()

# Adjust N for the interpolation by sps

data_src = numpy.array(put.vsnk_src.data())

# Convert the FLL's LO frequency from rads/sec to Hz

# adjust this to align with the data.

# Plot the Costas loop's LO frequency

s1.plot(data_frq)

s1.set_title("Costas LO")

s1.set_xlabel("Samples")

s1.set_ylabel("Frequency (normalized Hz)")

# Plot the IQ symbols

s3.plot(data_src.real, data_src.imag, "o")

s3.plot(data_cst.real, data_cst.imag, "rx")

s3.set_title("IQ")

s3.set_ylim([-2, 2])

# Plot the symbols in time

s4.set_position([0.125, 0.05, 0.775, 0.4])

s4.plot(data_src.real, "o-")

s4.plot(data_cst.real, "rx-")

pyplot.show()

if __name__ == "__main__":

try:

main()

except KeyboardInterrupt:

pass

print("Error: could not from matplotlib import pyplot (http://matplotlib.sourceforge.net/)")

sys.exit(1)

class example_fll(gr.top_block):

def __init__(self, N, sps, rolloff, ntaps, bw, noise, foffset, toffset, poffset):

gr.top_block.__init__(self)

rrc_taps = filter.firdes.root_raised_cosine(

sps, sps, 1.0, rolloff, ntaps)

self.src = blocks.vector_source_c(data.tolist(), False)

self.rrc = filter.interp_fir_filter_ccf(sps, rrc_taps)

self.connect(self.src, self.rrc, self.chn, self.fll, self.vsnk_fll)

self.connect(self.rrc, self.vsnk_src)

def main():

parser = ArgumentParser(conflict_handler="resolve")

parser.add_argument("-N", "--nsamples", type=int, default=2000,

parser.add_argument("-S", "--sps", type=int, default=4,

parser.add_argument("-r", "--rolloff", type=eng_float, default=0.35,

parser.add_argument("-n", "--ntaps", type=int, default=45,

parser.add_argument("--noise", type=eng_float, default=0.0,

parser.add_argument("-f", "--foffset", type=eng_float, default=0.2,

parser.add_argument("-t", "--toffset", type=eng_float, default=1.0,

parser.add_argument("-p", "--poffset", type=eng_float, default=0.0,

args = parser.parse_args()

# Adjust N for the interpolation by sps

data_err = numpy.array(put.vsnk_err.data())

# Convert the FLL's LO frequency from rads/sec to Hz

# adjust this to align with the data. There are 2 filters of

# ntaps long and the channel introduces another 4 sample delay.

# Plot the FLL's LO frequency

s1.plot(data_frq)

s1.set_title("FLL LO")

s1.set_xlabel("Samples")

s1.set_ylabel("Frequency (normalized Hz)")

# Plot the FLL's error

s2.plot(data_err)

s2.set_title("FLL Error")

s2.set_xlabel("Samples")

s2.set_ylabel("FLL Loop error")

# Plot the IQ symbols

s3.plot(data_src.real, data_src.imag, "o")

s3.plot(data_fll.real, data_fll.imag, "rx")

s3.set_title("IQ")

s3.set_ylabel("Imag part")

# Plot the symbols in time

s4.plot(data_src.real, "o-")

s4.plot(data_fll.real, "rx-")

s4.set_title("Symbols")

pyplot.show()

if __name__ == "__main__":

try:

main()

except KeyboardInterrupt:

pass

print("Error: could not from matplotlib import pyplot (http://matplotlib.sourceforge.net/)")

sys.exit(1)

class example_timing(gr.top_block):

def __init__(self, N, sps, rolloff, ntaps, bw, noise,

foffset, toffset, poffset, mode=0):

gain = bw

nfilts = 32

rrc_taps_rx = filter.firdes.root_raised_cosine(

self.src = blocks.vector_source_c(data.tolist(), False)

self.rrc = filter.interp_fir_filter_ccf(sps, rrc_taps)

if mode == 0:

self.clk = digital.pfb_clock_sync_ccf(sps, gain, rrc_taps_rx,

self.taps = self.clk.taps()

self.dtaps = self.clk.diff_taps()

self.vsnk_err = blocks.vector_sink_f()

self.vsnk_rat = blocks.vector_sink_f()

self.vsnk_phs = blocks.vector_sink_f()

mu = 0.5

gain_mu = bw

omega_rel_lim = 0.02

self.clk = digital.clock_recovery_mm_cc(sps, gain_omega,

mu, gain_mu,

self.vsnk_err = blocks.vector_sink_f()

self.vsnk_src = blocks.vector_sink_c()

self.vsnk_clk = blocks.vector_sink_c()

self.connect(self.src, self.vsnk_src)

def main():

parser = ArgumentParser(conflict_handler="resolve")

parser.add_argument("-N", "--nsamples", type=int, default=2000,

parser.add_argument("-S", "--sps", type=int, default=4,

parser.add_argument("-r", "--rolloff", type=eng_float, default=0.35,

parser.add_argument("-n", "--ntaps", type=int, default=45,

parser.add_argument("--noise", type=eng_float, default=0.0,

parser.add_argument("-f", "--foffset", type=eng_float, default=0.0,

parser.add_argument("-t", "--toffset", type=eng_float, default=1.0,

parser.add_argument("-p", "--poffset", type=eng_float, default=0.0,

parser.add_argument("-M", "--mode", type=int, default=0,

args = parser.parse_args()

# Adjust N for the interpolation by sps

data_rat = numpy.array(put.vsnk_rat.data()[20:])

data_phs = numpy.array(put.vsnk_phs.data()[20:])

# Plot the IQ symbols

s1.plot(data_src.real, data_src.imag, "bo")

s1.plot(data_clk.real, data_clk.imag, "ro")

s1.set_title("IQ")

# Plot the symbols in time

delay = put.delay

m = len(data_clk.real)

s2.plot(data_src.real, "bs", markersize=10, label="Input")

s2.plot(data_clk.real[delay:], "ro", label="Recovered")

s2.set_title("Symbols")

s2.legend()

# Plot the clock recovery loop's error

s3.plot(data_err, label="Error")

s3.plot(data_rat, 'r', label="Update rate")

s3.set_title("Clock Recovery Loop Error")

s3.legend()

# Plot the clock recovery loop's error

s4.plot(data_phs)

s4.set_title("Clock Recovery Loop Filter Phase")

s4.set_xlabel("Samples")

s4.set_ylabel("Filter Phase")

diff_taps = put.dtaps

ntaps = len(diff_taps[0])

nfilts = len(diff_taps)

s31.set_title("Differential Filters")

s32.set_title("FFT of Differential Filters")

s31.plot(t[i::nfilts].real, d, "-o")

s32.plot(D)

s32.set_ylim([-120, 10])

data_err = numpy.array(put.vsnk_err.data()[20:])

# Plot the IQ symbols

s1.plot(data_src.real, data_src.imag, "o")

s1.plot(data_clk.real, data_clk.imag, "ro")

s1.set_title("IQ")

s1.set_ylim([-2, 2])

# Plot the symbols in time

s2.plot(data_src.real, "bs", markersize=10, label="Input")

s2.plot(data_clk.real, "ro", label="Recovered")

s2.set_title("Symbols")

s2.legend()

# Plot the clock recovery loop's error

s3.plot(data_err)

s3.set_title("Clock Recovery Loop Error")

s3.set_xlabel("Samples")

pyplot.show()

if __name__ == "__main__":

try:

main()

#!/usr/bin/env python

#

# Copyright 2011,2013 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

from gnuradio import gr

from gnuradio import blocks

from argparse import ArgumentParser

import sys

class my_graph(gr.top_block):

def __init__(self):

self.dst = blocks.vector_sink_s()

self.connect(src, head, self.dst)

if __name__ == '__main__':

try:

tb = my_graph()

for s in tb.dst.data():

f.write("%3d, " % (s & 0xff,))

f.write("%3d, " % ((s >> 8) & 0xff,))

if i % 16 == 0:

f.write('\n')

except KeyboardInterrupt:

pass

from gnuradio.eng_option import eng_option

from optparse import OptionParser

class my_top_block(gr.top_block):

def __init__(self, ifile, ofile, options):

SNR = 10.0**(options.snr / 10.0)

frequency_offset = options.frequency_offset

time_offset = options.time_offset

# calculate noise voltage from SNR

power_in_signal = abs(options.tx_amplitude)**2

#self.throttle = blocks.throttle(gr.sizeof_gr_complex, options.sample_rate)

self.channel = channels.channel_model(noise_voltage, frequency_offset,

self.phase = blocks.multiply_const_cc(complex(math.cos(phase_offset),

self.snk = blocks.file_sink(gr.sizeof_gr_complex, ofile)

self.connect(self.src, self.channel, self.phase, self.snk)

def main():

# Create Options Parser:

usage = "benchmack_add_channel.py [options] <input file> <output file>"

parser.add_option("-n", "--snr", type="eng_float", default=30,

help="set the SNR of the channel in dB [default=%default]")

parser.add_option("", "--seed", action="store_true", default=False,

default=1.0,

help="tell the simulator the signal amplitude [default=%default]")

if len(args) != 2:

parser.print_help(sys.stderr)

tb.start() # start flow graph

tb.wait() # wait for it to finish

if __name__ == '__main__':

try:

main()

from optparse import OptionParser

from gnuradio.eng_option import eng_option

import threading

from gnuradio import digital

from gnuradio import blocks

n2s = eng_notation.num_to_str

class status_thread(threading.Thread):

def __init__(self, tb):

threading.Thread.__init__(self)

def run(self):

while not self.done:

print("Freq. Offset: {0:5.0f} Hz Timing Offset: {1:10.1f} ppm Estimated SNR: {2:4.1f} dB BER: {3:g}".format(

try:

time.sleep(1.0)

except KeyboardInterrupt:

self.done = True

class bert_receiver(gr.hier_block2):

def __init__(self, bitrate,

constellation, samples_per_symbol,

verbose, log):

gr.hier_block2.__init__(self, "bert_receive",

gr.io_signature(0, 0, 0)) # Output signature

self._bitrate = bitrate

self.connect(self._demod.time_recov, self._snr_probe)

# Descramble BERT sequence. A channel error will create 3 incorrect bits

# Measure BER by the density of 0s in the stream

self._ber = digital.probe_density_b(1.0 / self._symbol_rate)

self.connect(self, self._demod, self._descrambler, self._ber)

def frequency_offset(self):

def timing_offset(self):

return self._demod.time_recov.clock_rate()

return self._snr_probe.snr()

def ber(self):

class rx_psk_block(gr.top_block):

self._demodulator_class = demod

# Get demod_kwargs

# demodulator

self._demodulator = self._demodulator_class(**demod_kwargs)

options.samples_per_symbol = self._source._sps

elif(options.from_file is not None):

else:

self._source = blocks.null_source(gr.sizeof_gr_complex)

def get_options(demods):

parser = OptionParser(option_class=eng_option, conflict_handler="resolve")

help="input file of samples to demod")

parser.add_option("-m", "--modulation", type="choice", choices=list(demods.keys()),

default='psk',

parser.add_option("-S", "--samples-per-symbol", type="float", default=2,

help="set samples/symbol [default=%default]")

if not parser.has_option("--verbose"):

if not parser.has_option("--log"):

parser.add_option("", "--log", action="store_true", default=False,

uhd_receiver.add_options(parser)

if __name__ == "__main__":

will be removed or fundamentally reworked in a coming GNU Radio

release.""")

demods = digital.modulation_utils.type_1_demods()

n2s = eng_notation.num_to_str

class bert_transmit(gr.hier_block2):

def __init__(self, constellation, samples_per_symbol,

differential, excess_bw, gray_coded,

verbose, log):

gr.hier_block2.__init__(self, "bert_transmit",

# Create BERT data bit stream

self._mod = digital.generic_mod(constellation, differential,

samples_per_symbol,

gray_coded, excess_bw,

verbose, log)

self.connect(self._bits, self._scrambler, self._pack, self._mod, self)

options.samples_per_symbol = self._sink._sps

elif(options.to_file is not None):

else:

self._sink = blocks.null_sink(gr.sizeof_gr_complex)

self._transmitter = bert_transmit(self._modulator._constellation,

options.samples_per_symbol,

options.differential,

help="Select modulation bit rate (default=%default)")

parser.add_option("-S", "--samples-per-symbol", type="float", default=2,

help="set samples/symbol [default=%default]")

help="Output file for modulated samples")

if not parser.has_option("--verbose"):

if not parser.has_option("--log"):

parser.add_option("", "--log", action="store_true", default=False)

return (options, args)

if __name__ == "__main__":

will be removed or fundamentally reworked in a coming GNU Radio

release.""")

mods = digital.modulation_utils.type_1_mods()

import sys

def add_freq_option(parser):

"""

Hackery that has the -f / --freq option set both tx_freq and rx_freq

help="set Tx and/or Rx frequency to FREQ [default=%default]",

metavar="FREQ")

class uhd_interface(object):

def __init__(self, istx, args, sym_rate, sps, freq=None, lo_offset=None,

gain=None, spec=None, antenna=None, clock_source=None):

if(istx):

else:

# Set clock source

if(clock_source):

self.u.set_antenna(antenna, 0)

self._args = args

self._spec = spec

self._gain = self.set_gain(gain)

self._lo_offset = lo_offset

sps = actual_samp_rate / sym_rate

if(sps < 2):

else:

actual_sps = sps

break

if gain is None:

# if no gain was specified, use the mid-point in dB

g = self.u.get_gain_range()

print("\nNo gain specified.")

print("Setting gain to %f (from [%f, %f])" %

self.u.set_gain(gain, 0)

return gain

return freq

else:

frange = self.u.get_freq_range()

sys.exit(1)

#-------------------------------------------------------------------#

# TRANSMITTER

#-------------------------------------------------------------------#

class uhd_transmitter(uhd_interface, gr.hier_block2):

def __init__(self, args, sym_rate, sps, freq=None, lo_offset=None, gain=None,

spec=None, antenna=None, clock_source=None, verbose=False):

gr.hier_block2.__init__(self, "uhd_transmitter",

# Set up the UHD interface as a transmitter

uhd_interface.__init__(self, True, args, sym_rate, sps,

help="set transmit gain in dB (default is midpoint)")

parser.add_option("-C", "--clock-source", type="string", default=None,

help="select clock source (e.g. 'external') [default=%default]")

def _print_verbage(self):

"""

Prints information about the UHD transmitter

"""

print("\nUHD Transmitter:")

print("Gain: %f dB" % (self._gain))

print("Sample Rate: %ssps" % (eng_notation.num_to_str(self._rate)))

#-------------------------------------------------------------------#

# RECEIVER

def __init__(self, args, sym_rate, sps, freq=None, lo_offset=None, gain=None,

spec=None, antenna=None, clock_source=None, verbose=False):

gr.hier_block2.__init__(self, "uhd_receiver",

# Set up the UHD interface as a receiver

uhd_interface.__init__(self, False, args, sym_rate, sps,

parser.add_option("-C", "--clock-source", type="string", default=None,

help="select clock source (e.g. 'external') [default=%default]")

if not parser.has_option("--verbose"):

def _print_verbage(self):

"""

Prints information about the UHD transmitter

"""

print("\nUHD Receiver:")

print("Gain: %f dB" % (self._gain))

print("Sample Rate: %ssps" % (eng_notation.num_to_str(self._rate)))

from gnuradio.eng_option import eng_option

from optparse import OptionParser

class my_top_block(gr.top_block):

def __init__(self, ifile, ofile, options):

SNR = 10.0**(options.snr / 10.0)

time_offset = options.time_offset

# calculate noise voltage from SNR

power_in_signal = abs(options.tx_amplitude)**2

#self.throttle = blocks.throttle(gr.sizeof_gr_complex, options.sample_rate)

self.channel = channels.channel_model(noise_voltage, frequency_offset,

self.phase = blocks.multiply_const_cc(complex(math.cos(phase_offset),

self.snk = blocks.file_sink(gr.sizeof_gr_complex, ofile)

self.connect(self.src, self.channel, self.phase, self.snk)

def main():

# Create Options Parser:

usage = "benchmack_add_channel.py [options] <input file> <output file>"

parser.add_option("-n", "--snr", type="eng_float", default=30,

help="set the SNR of the channel in dB [default=%default]")

parser.add_option("", "--seed", action="store_true", default=False,

default=1.0,

help="tell the simulator the signal amplitude [default=%default]")

if len(args) != 2:

parser.print_help(sys.stderr)

tb.start() # start flow graph

tb.wait() # wait for it to finish

if __name__ == '__main__':

try:

main()

# receive path

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

class receive_path(gr.hier_block2):

def __init__(self, rx_callback, options):

gr.io_signature(1, 1, gr.sizeof_gr_complex),

gr.io_signature(0, 0, 0))

# receiver

self.ofdm_rx = digital.ofdm_demod(options,

# Carrier Sensing Blocks

alpha = 0.001

thresh = 30 # in dB, will have to adjust

self.connect(self, self.ofdm_rx)

self.connect(self.ofdm_rx, self.probe)

"""

Return True if we think carrier is present.

"""

return self.probe.unmuted()

def carrier_threshold(self):

normal.add_option("-W", "--bandwidth", type="eng_float",

default=500e3,

help="set symbol bandwidth [default=%default]")

expert.add_option("", "--log", action="store_true", default=False,

help="Log all parts of flow graph to files (CAUTION: lots of data)")

# transmit path

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

class transmit_path(gr.hier_block2):

def __init__(self, options):

'''

gr.io_signature(0, 0, 0),

gr.io_signature(1, 1, gr.sizeof_gr_complex))

self.ofdm_tx = digital.ofdm_mod(options,

msgq_limit=4,

Prints information about the transmit path

"""

print("Tx amplitude %s" % (self._tx_amplitude))

import sys

def add_freq_option(parser):

"""

Hackery that has the -f / --freq option set both tx_freq and rx_freq

help="set Tx and/or Rx frequency to FREQ [default=%default]",

metavar="FREQ")

class uhd_interface(object):

def __init__(self, istx, args, bandwidth, freq=None, lo_offset=None,

gain=None, spec=None, antenna=None, clock_source=None):

if(istx):

else:

# Set clock source to external.

if(clock_source):

self.u.set_antenna(antenna, 0)

self._args = args

self._spec = spec

self._gain = self.set_gain(gain)

self._lo_offset = lo_offset

if gain is None:

# if no gain was specified, use the mid-point in dB

g = self.u.get_gain_range()

print("\nNo gain specified.")

print("Setting gain to %f (from [%f, %f])" %

self.u.set_gain(gain, 0)

return gain

return freq

else:

frange = self.u.get_freq_range()

sys.exit(1)

#-------------------------------------------------------------------#

# TRANSMITTER

#-------------------------------------------------------------------#

class uhd_transmitter(uhd_interface, gr.hier_block2):

def __init__(self, args, bandwidth, freq=None, lo_offset=None, gain=None,

spec=None, antenna=None, clock_source=None, verbose=False):

gr.hier_block2.__init__(self, "uhd_transmitter",

# Set up the UHD interface as a transmitter

uhd_interface.__init__(self, True, args, bandwidth,

help="set transmit gain in dB (default is midpoint)")

parser.add_option("-C", "--clock-source", type="string", default=None,

help="select clock source (e.g. 'external') [default=%default]")

def _print_verbage(self):

"""

Prints information about the UHD transmitter

"""

print("\nUHD Transmitter:")

print("Gain: %f dB" % (self._gain))

print("Sample Rate: %ssps" % (eng_notation.num_to_str(self._rate)))

#-------------------------------------------------------------------#

def __init__(self, args, bandwidth, freq=None, lo_offset=None, gain=None,

spec=None, antenna=None, clock_source=None, verbose=False):

gr.hier_block2.__init__(self, "uhd_receiver",

# Set up the UHD interface as a receiver

uhd_interface.__init__(self, False, args, bandwidth,

parser.add_option("-C", "--clock-source", type="string", default=None,

help="select clock source (e.g. 'external') [default=%default]")

if not parser.has_option("--verbose"):

def _print_verbage(self):

"""

Prints information about the UHD transmitter

"""

print("\nUHD Receiver:")

print("Gain: %f dB" % (self._gain))

print("Sample Rate: %ssps" % (eng_notation.num_to_str(self._rate)))

from optparse import OptionParser

import sys

def main():

parser = OptionParser()

parser.add_option("-f", "--file", default=None,

for ch in f.read():

x = ord(ch)

bytes = bytes + 1

bits = bits + 1

t = (x >> i) & 0x1

if t == current:

if count > 0:

if len(runs) < count:

for j in range(count - len(runs)):

count = 1

# Deal with last run at EOF

if len(runs) < count and count > 0:

for j in range(count - len(runs)):

chk = 0

print("Bytes read: ", bytes)

print("Bits read: ", bits)

print()

for i in range(len(runs)):

print()

print("Sum of runs:", chk, "bits")

print()

print("Maximum run length is", len(runs), "bits")

if __name__ == "__main__":

main()

http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Higher-order_statistics

'''

def online_skewness(data):

n = 0

mean = 0

for n in range(len(data)):

delta = data[n] - mean

term1 = delta * delta_n * n

mean = mean + delta_n

M3 = M3 + term1 * delta_n * (n - 1) - 3 * delta_n * M2

M2 = M2 + term1

def snr_est_simple(signal):

s = scipy.mean(abs(signal)**2)

snr_rat = s / n

def snr_est_skew(signal):

y1 = scipy.mean(abs(signal))

y2 = scipy.mean(scipy.real(signal**2))

y4 = online_skewness(signal.real)

#y4 = stats.skew(abs(signal.real))

snr_rat = s / n

def snr_est_m2m4(signal):

M2 = scipy.mean(abs(signal)**2)

M4 = scipy.mean(abs(signal)**4)

def snr_est_svr(signal):

N = len(signal)

ssum = 0

msum = 0

for i in range(1, N):

msum += (abs(signal[i])**4)

beta = savg / (mavg - savg)

def main():

"m2m4": snr_est_m2m4,

"svr": snr_est_svr}

parser = OptionParser(option_class=eng_option, conflict_handler="resolve")

parser.add_option("-N", "--nsamples", type="int", default=10000,

help="Set the number of samples to process [default=%default]")

choices=list(gr_estimators.keys()), default="simple",

help="Estimator type {0} [default=%default]".format(

list(gr_estimators.keys())))

N = options.nsamples

xx = scipy.random.randn(N)

xy = scipy.random.randn(N)

# 1j*(2*scipy.complex64(scipy.random.randint(0, 2, N)) - 1)

snr_known = list()

# when to issue an SNR tag; can be ignored in this example.

ntag = 10000

py_est = py_estimators[options.type]

gr_est = gr_estimators[options.type]

SNR_min = options.snr_min

SNR_max = options.snr_max

SNR_step = options.snr_step

for snr in SNR_dB:

SNR = 10.0**(snr / 10.0)

yy = bits + n_cpx / scale

print("SNR: ", snr)

Sknown = scipy.mean(yy**2)

Nknown = scipy.var(n_cpx / scale)

snr0 = Sknown / Nknown

snr_known.append(float(snr0dB))

snrdB, snr = py_est(yy)

snr_gr.append(gr_snr.snr())

f1 = pyplot.figure(1)

s1.plot(SNR_dB, snr_known, "k-o", linewidth=2, label="Known")

s1.plot(SNR_dB, snr_python, "b-o", linewidth=2, label="Python")

s1.plot(SNR_dB, snr_gr, "g-o", linewidth=2, label="GNU Radio")

s1.legend()

f2 = pyplot.figure(2)

s2.plot(yy.real, yy.imag, 'o')

pyplot.show()

__path__.append(os.path.join(dirname, "bindings"))

from .digital_python import *

from .psk import *

from .qam import *

from .qamlike import *

from .qam_constellations import *

from .constellation_map_generator import *

# BPSK constellation

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

def bpsk_constellation():

return digital_python.constellation_bpsk()

# DBPSK constellation

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

def dbpsk_constellation():

return digital_python.constellation_dbpsk()

symbols = list()

for s_i in s:

tmp = 0

bit = (s_i >> i) & 0x1

tmp |= bit << p

symbols.append(tmp ^ k)

_def_bits_per_symbol = 1

_def_h_numerator = 1

_def_h_denominator = 2

_def_bt = 0.35

_def_symbols_per_pulse = 1

_def_generic_taps = numpy.empty(1)

log=_def_log):

gr.hier_block2.__init__(self, "cpm_mod",

self._samples_per_symbol = samples_per_symbol

self._bits_per_symbol = bits_per_symbol

self._h_numerator = h_numerator

self._h_denominator = h_denominator

self._cpm_type = cpm_type

self._symbols_per_pulse = symbols_per_pulse

self._symbols_per_pulse = 4

else:

if samples_per_symbol < 2:

self.nsymbols = 2**bits_per_symbol

self.ntaps = int(self._symbols_per_pulse * samples_per_symbol)

sensitivity = 2 * pi * h_numerator / h_denominator / samples_per_symbol

# Unpack Bytes into bits_per_symbol groups

# Turn it into symmetric PAM data.

# Generate pulse (sum of taps = samples_per_symbol/2)

gaussian_taps = filter.firdes.gaussian(

1.0 / 2, # gain

samples_per_symbol, # symbol_rate

bt, # bandwidth * symbol time

self.ntaps # number of taps

sqwave = (1,) * samples_per_symbol # rectangular window

# generalize it for arbitrary roll-off factor

self.taps = generic_taps

else:

# FM modulation

self.fmmod = analog.frequency_modulator_fc(sensitivity)

def symbols_per_pulse(self):

return self._symbols_per_pulse

def _print_verbage(self):

print("Samples per symbol = %d" % self._samples_per_symbol)

print("Bits per symbol = %d" % self._bits_per_symbol)

print("Symbols per pulse = %d" % self._symbols_per_pulse)

print("CPM type = %d" % self._cpm_type)

if self._cpm_type == 1:

def _setup_logging(self):

print("Modulation logging turned on.")

Given command line options, create dictionary suitable for passing to __init__

"""

return modulation_utils.extract_kwargs_from_options(cpm_mod.__init__,

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

#

# Not yet implemented

#

#

# Add these to the mod/demod registry

#

_def_truncate = False

# Frequency correction

# Symbol timing recovery

_def_timing_max_dev = 1.5

# Fine frequency / Phase correction

# Number of points in constellation

_def_constellation_points = 16

# Whether differential coding is used.

_def_differential = False

def add_common_options(parser):

"""

Sets options common to both modulator and demodulator.

parser.add_option("", "--mod-code", type="choice", choices=mod_codes.codes,

default=mod_codes.NO_CODE,

help="Select modulation code from: %s [default=%%default]"

parser.add_option("", "--excess-bw", type="float", default=_def_excess_bw,

help="set RRC excess bandwidth factor [default=%default]")

truncate=_def_truncate):

gr.hier_block2.__init__(self, "generic_mod",

self._constellation = constellation

self._samples_per_symbol = samples_per_symbol

self.pre_diff_code = pre_diff_code and self._constellation.apply_pre_diff_code()

if self._samples_per_symbol < 2:

# turn bytes into k-bit vectors

self.bytes2chunks = \

if self.pre_diff_code:

if differential:

self.diffenc = digital.diff_encoder_bb(arity)

# pulse shaping filter

nfilts = 32

ntaps_per_filt = 11

self.rrc_taps = filter.firdes.root_raised_cosine(

nfilts, # gain

nfilts, # sampling rate based on 32 filters in resampler

1.0, # symbol rate

ntaps)

self.rrc_filter = filter.pfb_arb_resampler_ccf(self._samples_per_symbol,

self.rrc_taps)

# Remove the filter transient at the beginning of the transmission

if truncate:

fsps = float(self._samples_per_symbol)

# Connect

self._blocks = [self, self.bytes2chunks]

if differential:

self._blocks.append(self.diffenc)

self._blocks += [self.chunks2symbols, self.rrc_filter]

if truncate:

self._blocks.append(self.skiphead)

self._blocks.append(self)

if log:

self._setup_logging()

def samples_per_symbol(self):

return self._samples_per_symbol

Given command line options, create dictionary suitable for passing to __init__

"""

return extract_kwargs_from_options_for_class(cls, options)

def _print_verbage(self):

print("\nModulator:")

log=_def_log):

gr.hier_block2.__init__(self, "generic_demod",

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

self._constellation = constellation

self._phase_bw = phase_bw

self._freq_bw = freq_bw

self._timing_bw = timing_bw

self._differential = differential

if self._samples_per_symbol < 2:

# Only apply a predifferential coding if the constellation also supports it.

self.pre_diff_code = pre_diff_code and self._constellation.apply_pre_diff_code()

nfilts = 32

# Automatic gain control

self.agc = analog.agc2_cc(0.6e-1, 1e-3, 1, 1)

fll_ntaps, self._freq_bw)

# symbol timing recovery with RRC data filter

1.0, self._excess_bw, ntaps)

self.time_recov = digital.pfb_clock_sync_ccf(self._samples_per_symbol,

self._timing_bw, taps,

fmin = -0.25

fmax = 0.25

def _print_verbage(self):

print("\nDemodulator:")

print("RRC roll-off factor: %.2f" % self._excess_bw)

print("FLL bandwidth: %.2e" % self._freq_bw)

print("Timing bandwidth: %.2e" % self._timing_bw)

Given command line options, create dictionary suitable for passing to __init__

"""

return extract_kwargs_from_options_for_class(cls, options)

shared_demod_args = """ samples_per_symbol: samples per baud >= 2 (float)

excess_bw: Root-raised cosine filter excess bandwidth (float)

do_unpack=_def_do_unpack):

gr.hier_block2.__init__(self, "gfsk_mod",

samples_per_symbol = int(samples_per_symbol)

self._samples_per_symbol = samples_per_symbol

self._differential = False

if not isinstance(samples_per_symbol, int) or samples_per_symbol < 2:

ntaps = 4 * samples_per_symbol # up to 3 bits in filter at once

# Turn it into NRZ data.

#self.nrz = digital.bytes_to_syms()

# Form Gaussian filter

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

self.gaussian_taps = filter.firdes.gaussian(

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

# FM modulation

self.fmmod = analog.frequency_modulator_fc(sensitivity)

# Connect & Initialize base class

if do_unpack:

self.unpack = blocks.packed_to_unpacked_bb(1, gr.GR_MSB_FIRST)

else:

def samples_per_symbol(self):

return self._samples_per_symbol

@staticmethod

return 1

def _print_verbage(self):

print("bits per symbol = %d" % self.bits_per_symbol())

print("Gaussian filter bt = %.2f" % self._bt)

def _setup_logging(self):

print("Modulation logging turned on.")

self.connect(self.nrz,

log=_def_log):

gr.hier_block2.__init__(self, "gfsk_demod",

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

self._samples_per_symbol = samples_per_symbol

self._differential = False

if samples_per_symbol < 2:

if not self._gain_mu:

self._gain_mu = 0.175

self._damping = 1.0

self._max_dev = self._omega_relative_limit * self._samples_per_symbol

# Demodulate FM

# the clock recovery block tracks the symbol clock and resamples as needed.

# the output of the block is a stream of soft symbols (float)

self.clock_recovery = self.digital_symbol_sync_xx_0 = digital.symbol_sync_ff(digital.TED_MUELLER_AND_MULLER,

# slice the floats at 0, outputting 1 bit (the LSB of the output byte) per sample

self.slicer = digital.binary_slicer_fb()

self._setup_logging()

# Connect & Initialize base class

def samples_per_symbol(self):

return self._samples_per_symbol

print("Symbol Sync M&M omega = %f" % self._omega)

print("Symbol Sync M&M gain mu = %f" % self._gain_mu)

print("M&M clock recovery mu (Unused) = %f" % self._mu)

print("frequency error = %f" % self._freq_error)

def _setup_logging(self):

print("Demodulation logging turned on.")

self.connect(self.fmdemod,

self.connect(self.clock_recovery,

self.connect(self.slicer,

@staticmethod

def add_options(parser):

do_unpack=_def_do_unpack):

gr.hier_block2.__init__(self, "gmsk_mod",

samples_per_symbol = int(samples_per_symbol)

self._samples_per_symbol = samples_per_symbol

self._differential = False

if not isinstance(samples_per_symbol, int) or samples_per_symbol < 2:

# Turn it into NRZ data.

#self.nrz = digital.bytes_to_syms()

# Form Gaussian filter

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

self.gaussian_taps = filter.firdes.gaussian(

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

# FM modulation

self.fmmod = analog.frequency_modulator_fc(sensitivity)

# Connect & Initialize base class

if do_unpack:

self.unpack = blocks.packed_to_unpacked_bb(1, gr.GR_MSB_FIRST)

else:

def samples_per_symbol(self):

return self._samples_per_symbol

@staticmethod

return 1

def _print_verbage(self):

print("bits per symbol = %d" % self.bits_per_symbol())

print("Gaussian filter bt = %.2f" % self._bt)

def _setup_logging(self):

print("Modulation logging turned on.")

self.connect(self.nrz,

log=_def_log):

gr.hier_block2.__init__(self, "gmsk_demod",

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

self._samples_per_symbol = samples_per_symbol

self._differential = False

if samples_per_symbol < 2:

if not self._gain_mu:

self._gain_mu = 0.175

self._damping = 1.0

self._max_dev = self._omega_relative_limit * self._samples_per_symbol

# Demodulate FM

# the clock recovery block tracks the symbol clock and resamples as needed.

# the output of the block is a stream of soft symbols (float)

self.clock_recovery = self.digital_symbol_sync_xx_0 = digital.symbol_sync_ff(digital.TED_MUELLER_AND_MULLER,

# slice the floats at 0, outputting 1 bit (the LSB of the output byte) per sample

self.slicer = digital.binary_slicer_fb()

self._setup_logging()

# Connect & Initialize base class

def samples_per_symbol(self):

return self._samples_per_symbol

print("Symbol Sync M&M omega = %f" % self._omega)

print("Symbol Sync M&M gain mu = %f" % self._gain_mu)

print("M&M clock recovery mu (Unused) = %f" % self._mu)

print("frequency error = %f" % self._freq_error)

def _setup_logging(self):

print("Demodulation logging turned on.")

self.connect(self.fmdemod,

self.connect(self.clock_recovery,

self.connect(self.slicer,

@staticmethod

def add_options(parser):

#

# Copyright 2010 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

# Type 1 modulators accept a stream of bytes on their input and produce complex baseband output

_type_1_modulators = {}

def type_1_mods():

return _type_1_modulators

def add_type_1_mod(name, mod_class):

_type_1_modulators[name] = mod_class

# to resolve phase or polarity ambiguities.

_type_1_demodulators = {}

def type_1_demods():

return _type_1_demodulators

def add_type_1_demod(name, demod_class):

_type_1_demodulators[name] = demod_class

# Also record the constellation making functions of the modulations

_type_1_constellations = {}

def type_1_constellations():

return _type_1_constellations

def add_type_1_constellation(name, constellation):

_type_1_constellations[name] = constellation

excluded_args: function arguments that are NOT to be added to the dictionary (sequence of strings)

options: result of command argument parsing (optparse.Values)

"""

# Try this in C++ ;)

spec = inspect.getfullargspec(function)

d = {}

d[kw] = getattr(options, kw)

return d

def extract_kwargs_from_options_for_class(cls, options):

"""

Given command line options, create dictionary suitable for passing to __init__

_def_packet_length_tag_key = "packet_length"

_def_packet_num_tag_key = "packet_num"

# Data and pilot carriers are same as in 802.11a

_pilot_sym_scramble_seq = (

)

_seq_seed = 42

active_carriers.append(carrier)

return active_carriers

def _make_sync_word1(fft_len, occupied_carriers, pilot_carriers):

""" Creates a random sync sequence for fine frequency offset and timing

estimation. This is the first of typically two sync preamble symbols

Carrier 0 (DC carrier) is always zero. If used, carrier 1 is non-zero.

This means the sync algorithm has to check on odd carriers!

"""

numpy.random.seed(_seq_seed)

bpsk = {0: numpy.sqrt(2), 1: -numpy.sqrt(2)}

return numpy.fft.fftshift(sw1)

def _make_sync_word2(fft_len, occupied_carriers, pilot_carriers):

""" Creates a random sync sequence for coarse frequency offset and channel

estimation. This is the second of typically two sync preamble symbols

for the Schmidl & Cox sync algorithm.

Symbols are always BPSK symbols.

"""

numpy.random.seed(_seq_seed)

bpsk = {0: 1, 1: -1}

sw2[0] = 0j

return numpy.fft.fftshift(sw2)

def _get_constellation(bps):

""" Returns a modulator block for a given number of bits per symbol """

constellation = {

}

try:

return constellation[bps]

print('Modulation not supported.')

exit(1)

class ofdm_tx(gr.hier_block2):

"""Hierarchical block for OFDM modulation.

scramble_bits: Activates the scramblers (set this to True unless debugging)

"""

def __init__(self, fft_len=_def_fft_len, cp_len=_def_cp_len,

packet_length_tag_key=_def_packet_length_tag_key,

occupied_carriers=_def_occupied_carriers,

scramble_bits=False

):

gr.hier_block2.__init__(self, "ofdm_tx",

### Param init / sanity check ########################################

self.packet_length_tag_key = packet_length_tag_key

self.occupied_carriers = occupied_carriers

self.sync_word1 = sync_word1

if sync_word1 is None:

else:

if len(sync_word1) != self.fft_len:

if sync_word2 is None:

else:

self.sync_word2 = sync_word2

if len(self.sync_word2):

if len(self.sync_word2) != fft_len:

self.sync_word2 = list(self.sync_word2)

self.sync_words.append(self.sync_word2)

if scramble_bits:

self.scramble_seed = 0x7f

else:

### Header modulation ################################################

crc = digital.crc32_bb(False, self.packet_length_tag_key)

formatter_object = digital.packet_header_ofdm(

occupied_carriers=occupied_carriers, n_syms=1,

bits_per_header_sym=self.bps_header,

bits_per_payload_sym=self.bps_payload,

scramble_header=scramble_bits

)

header_payload_mux = blocks.tagged_stream_mux(

)

self.connect(

)

if debug_log:

self.connect(header_gen, blocks.file_sink(1, 'tx-hdr.dat'))

### Payload modulation ###############################################

payload_constellation = _get_constellation(bps_payload)

payload_scrambler = digital.additive_scrambler_bb(

0x8a,

self.scramble_seed,

7,

reset_tag_key=self.packet_length_tag_key

)

payload_unpack = blocks.repack_bits_bb(

bps_payload,

self.packet_length_tag_key

)

len_tag_key=self.packet_length_tag_key

)

ffter = fft.fft_vcc(

)

cyclic_prefixer = digital.ofdm_cyclic_prefixer(

self.fft_len,

rolloff,

self.packet_length_tag_key

)

if debug_log:

class ofdm_rx(gr.hier_block2):

| entirely. Also used for coarse frequency offset and

| channel estimation.

"""

def __init__(self, fft_len=_def_fft_len, cp_len=_def_cp_len,

frame_length_tag_key=_def_frame_length_tag_key,

packet_length_tag_key=_def_packet_length_tag_key,

scramble_bits=False

):

gr.hier_block2.__init__(self, "ofdm_rx",

### Param init / sanity check ########################################

self.occupied_carriers = occupied_carriers

n_sync_words = 1

if sync_word1 is None:

else:

if len(sync_word1) != self.fft_len:

self.sync_word1 = sync_word1

self.sync_word2 = ()

if sync_word2 is None:

n_sync_words = 2

elif len(sync_word2):

if len(sync_word2) != fft_len:

self.sync_word2 = sync_word2

n_sync_words = 2

if scramble_bits:

self.scramble_seed = 0x7f

else:

### Sync ############################################################

sync_detect = digital.ofdm_sync_sc_cfb(fft_len, cp_len)

oscillator = analog.frequency_modulator_fc(-2.0 / fft_len)

mixer = blocks.multiply_cc()

hpd = digital.header_payload_demux(

fft_len, cp_len, # FFT length, guard interval

"", # We're not using trigger tags

)

self.connect(self, sync_detect)

self.connect(self, delay, (mixer, 0), (hpd, 0))

self.connect((sync_detect, 0), oscillator, (mixer, 1))

self.connect((sync_detect, 1), (hpd, 1))

if debug_log:

### Header demodulation ##############################################

header_constellation = _get_constellation(bps_header)

fft_len,

header_constellation.base(),

occupied_carriers,

symbols_skipped=0,

)

header_eq = digital.ofdm_frame_equalizer_vcvc(

)

header_serializer = digital.ofdm_serializer_vcc(

)

header_formatter = digital.packet_header_ofdm(

)

self.connect(

)

self.msg_connect(header_parser, "header_data", hpd, "header_data")

if debug_log:

### Payload demod ####################################################

payload_fft = fft.fft_vcc(self.fft_len, True, (), True)

payload_constellation = _get_constellation(bps_payload)

payload_equalizer = digital.ofdm_equalizer_simpledfe(

)

payload_eq = digital.ofdm_frame_equalizer_vcvc(

)

payload_serializer = digital.ofdm_serializer_vcc(

)

self.payload_descrambler = digital.additive_scrambler_bb(

0x8a,

self.scramble_seed,

7,

reset_tag_key=self.packet_length_tag_key

)

self.crc = digital.crc32_bb(True, self.packet_length_tag_key)

self.connect(

)

if debug_log:

# Default use of differential encoding

_def_differential = True

def create_encodings(mod_code, arity, differential):

post_diff_code = None

if mod_code not in mod_codes.codes:

pre_diff_code = []

post_diff_code = None

else:

return (pre_diff_code, post_diff_code)

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

# PSK constellation

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

def psk_constellation(m=_def_constellation_points, mod_code=_def_mod_code,

differential=_def_differential):

"""

"""

k = log(m) / log(2.0)

if (k != int(k)):

pre_diff_code, post_diff_code = create_encodings(mod_code, m, differential)

if post_diff_code is not None:

inverse_post_diff_code = mod_codes.invert_code(post_diff_code)

# PSK modulator

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

class psk_mod(generic_mod):

"""

Hierarchical block for RRC-filtered PSK modulation.

mod_code=_def_mod_code,

differential=_def_differential,

*args, **kwargs):

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

# PSK demodulator

#

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

class psk_demod(generic_demod):

"""

"""

# See generic_mod for additional arguments

__doc__ += shared_mod_args

def __init__(self, constellation_points=_def_constellation_points,

mod_code=_def_mod_code,

differential=_def_differential,

*args, **kwargs):

#

# Add these to the mod/demod registry

# BPSK Constellation Mappings

def psk_2_0x0():

'''

0 | 1

const_points = [-1, 1]

symbols = [0, 1]

return (const_points, symbols)

psk_2 = psk_2_0x0 # Basic BPSK rotation

psk_2_0 = psk_2 # First ID for BPSK rotations

def psk_2_0x1():

'''

1 | 0

const_points = [-1, 1]

symbols = [1, 0]

return (const_points, symbols)

psk_2_1 = psk_2_0x1

0 | 1

'''

x_re = x.real

sd_psk_2 = sd_psk_2_0x0 # Basic BPSK rotation

sd_psk_2_0 = sd_psk_2 # First ID for BPSK rotations

def sd_psk_2_0x1(x, Es=1):

'''

1 | 0

'''

sd_psk_2_1 = sd_psk_2_0x1

| -------

| 00 | 01

'''

symbols = [0, 1, 2, 3]

return (const_points, symbols)

psk_4 = psk_4_0x0_0_1

psk_4_0 = psk_4

def psk_4_0x1_0_1():

'''

| 11 | 10

k = 0x1

pi = [0, 1]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_1 = psk_4_0x1_0_1

def psk_4_0x2_0_1():

'''

| 00 | 01

k = 0x2

pi = [0, 1]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_2 = psk_4_0x2_0_1

def psk_4_0x3_0_1():

'''

| 01 | 00

k = 0x3

pi = [0, 1]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_3 = psk_4_0x3_0_1

def psk_4_0x0_1_0():

'''

| 01 | 11

k = 0x0

pi = [1, 0]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_4 = psk_4_0x0_1_0

def psk_4_0x1_1_0():

'''

| 00 | 10

k = 0x1

pi = [1, 0]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_5 = psk_4_0x1_1_0

def psk_4_0x2_1_0():

'''

| 11 | 01

k = 0x2

pi = [1, 0]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

psk_4_6 = psk_4_0x2_1_0

def psk_4_0x3_1_0():

'''

| 10 | 00

k = 0x3

pi = [1, 0]

return constellation_map_generator(psk_4()[0], psk_4()[1], k, pi)

############################################################

# QPSK Constellation Softbit LUT generators

'''

x_re = x.real

x_im = x.imag

sd_psk_4 = sd_psk_4_0x0_0_1

sd_psk_4_0 = sd_psk_4

def sd_psk_4_0x1_0_1(x, Es=1):

'''

| 11 | 10

'''

x_re = x.real

x_im = x.imag

sd_psk_4_1 = sd_psk_4_0x1_0_1

def sd_psk_4_0x2_0_1(x, Es=1):

'''

| 00 | 01

'''

x_re = x.real

x_im = x.imag

sd_psk_4_2 = sd_psk_4_0x2_0_1

def sd_psk_4_0x3_0_1(x, Es=1):

'''

| 01 | 00

'''

x_re = x.real

x_im = x.imag

sd_psk_4_3 = sd_psk_4_0x3_0_1

def sd_psk_4_0x0_1_0(x, Es=1):

'''

| 01 | 11

'''

x_re = x.real

x_im = x.imag

sd_psk_4_4 = sd_psk_4_0x0_1_0

def sd_psk_4_0x1_1_0(x, Es=1):

'''

| 00 | 10

'''

x_re = x.real

x_im = x.imag

sd_psk_4_5 = sd_psk_4_0x1_1_0

'''

x_re = x.real

x_im = x.imag

sd_psk_4_6 = sd_psk_4_0x2_1_0

def sd_psk_4_0x3_1_0(x, Es=1):

'''

| 10 | 00

'''

x_re = x.real

x_im = x.imag

sd_psk_4_7 = sd_psk_4_0x3_1_0

-4.0 * np.ones(5, dtype=complex)))

tags = (make_length_tag(0, length),)

expected = np.concatenate((np.zeros(prepad, dtype=complex), window[0:5],

window[5:10], np.zeros(postpad,

dtype=complex)))

etag = make_length_tag(0, length + prepad + postpad)

window = np.concatenate((-2.0 * np.ones(5), -4.0 * np.ones(5)))

tags = (make_length_tag(0, length1), make_length_tag(length1, length2))

expected = np.concatenate((np.zeros(prepad), window[0:5],

-1.0 * np.ones(length2 - len(window)),

-1.0 * window[5:10], np.zeros(postpad)))

etags = (make_length_tag(0, length1 + prepad + postpad),

make_tag(tag4_offset, 'body', pmt.intern('tag4')),

make_tag(tag5_offset, 'body', pmt.intern('tag5')))

expected = np.concatenate((np.zeros(prepad), window[0:5],

-1.0 * np.ones(length2 - len(window)),

-1.0 * window[5:10], np.zeros(postpad)))

elentag1_offset = 0

data[first:],

msg=msg)

def test_soft_qpsk_gen(self):

prec = 8

constel, code = digital.psk_4_0()

from gnuradio import gr, gr_unittest, digital, blocks

import numpy as np

class test_constellation_encoder(gr_unittest.TestCase):

def setUp(self):

def test_constellation_encoder_bc_bpsk(self):

cnst = digital.constellation_bpsk()

const_map = [-1.0, 1.0]

expected_result = [const_map[x] for x in src_data]

def test_constellation_encoder_bc_qpsk(self):

cnst = digital.constellation_qpsk()

expected_result = [cnst.points()[x] for x in src_data]

src = blocks.vector_source_b(src_data)

op = digital.constellation_encoder_bc(cnst.base())

# print "expected result", expected_result

self.assertFloatTuplesAlmostEqual(expected_result, actual_result)

def test_constellation_encoder_bc_qpsk_random(self):

cnst = digital.constellation_qpsk()

src_data = np.random.randint(0, 4, size=20000)

# print "expected result", expected_result

self.assertFloatTuplesAlmostEqual(expected_result, actual_result)

if __name__ == '__main__':

gr_unittest.run(test_constellation_encoder)

1,

1,

alg,

[],

'')

dst = blocks.vector_sink_c()

lambda: digital.glfsr_source_f(0))

self.assertRaises(RuntimeError,

lambda: digital.glfsr_source_f(65))

def test_005_correlation_f(self):

for degree in range(

1, 11): # Higher degrees take too long to correlate

from gnuradio import gr, gr_unittest, digital

from gnuradio.digital.utils import lfsr_args

class test_lfsr(gr_unittest.TestCase):

def setUp(self):

self.assertFloatTuplesAlmostEqual(expected_result, result_data, 5)

def test_lfsr_002(self):

expected_result = [1, 0, 0, 0, 0, 1, 0, 1, 0, 1, 1, 1, 0, 1, 1, 0,

self.assertEqual(expected_result, result_data)

if __name__ == '__main__':

gr_unittest.run(test_lfsr)

4,

1,

alg,

[],

'')

dst = blocks.vector_sink_c()

for x in range(fft_len)]

src = blocks.vector_source_c(tx_data, False, fft_len)

chan = blocks.multiply_const_vcc(channel)

add = blocks.add_cc(fft_len)

chanest = digital.ofdm_chanest_vcvc(sync_sym1, sync_sym2, 1)

sink = blocks.vector_sink_c(fft_len)

shifted_carrier_mask = shift_tuple(carrier_mask, carr_offset)

for i in range(fft_len):

if shifted_carrier_mask[i] and channel_est[i]:

rx_sym_est[i] = (sink.data()[i] / channel_est[i]).real

return carr_offset, list(shift_tuple(rx_sym_est, -carr_offset_hat))

bit_errors = 0

for _ in range(n_bursts):

gap = [0, ] * random.randint(0, 2 * fft_len)

tx_signal += gap + \

# Very loose definition of SNR here

snr = 20 # dB

sigma = 10**(-snr / 10)

out.append(d ^ scramble_word)

return out

class test_scrambler(gr_unittest.TestCase):

def setUp(self):

def tearDown(self):

self.tb = None

def test_lfsr_002(self):

l = digital.lfsr(*_a)

seq = [l.next_bit() for _ in range(2**10)]

def test_scrambler_descrambler_001(self):

src = blocks.vector_source_b(src_data, False)

m_tap = blocks.vector_sink_b()

dst = blocks.vector_sink_b()

self.tb.connect(src, scrambler, descrambler, dst)

self.tb.connect(scrambler, m_tap)

self.tb.run()

def test_scrambler_descrambler_002(self):

src = blocks.vector_source_b(src_data, False)

scrambler = digital.scrambler_bb(*_a)

m_tap = blocks.vector_sink_b()

self.tb.connect(src, scrambler, descrambler, dst)

self.tb.connect(scrambler, m_tap)

self.tb.run()

def test_scrambler_descrambler_003(self):

src = blocks.vector_source_b(src_data, False)

dst = blocks.vector_sink_b()

self.tb.connect(src, scrambler, descrambler1, descrambler2, dst)

self.tb.run()

def test_additive_scrambler_001(self):

src = blocks.vector_source_b(src_data, False)

scrambler = digital.additive_scrambler_bb(*_a)

descrambler = digital.additive_scrambler_bb(*_a)

self.assertEqual(tuple(src_data), tuple(dst.data()))

def test_additive_scrambler_002(self):

src = blocks.vector_source_b(src_data, False)

scrambler = digital.additive_scrambler_bb(*_a)

dst = blocks.vector_sink_b()

self.tb.connect(src, scrambler, dst)

self.tb.run()

def test_scrambler_descrambler(self):

src_data = [1, ] * 1000

# use the center point.

c = ((real_x - side / 2.0 + 0.5) * width +

(imag_x - side / 2.0 + 0.5) * width * 1j)

# This is not an edge row/column. Find closest point.

index = find_closest_point(c, points)

else:

# QPSK constellation

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

def qpsk_constellation(mod_code=_def_mod_code):

"""

Creates a QPSK constellation.

"""

if mod_code != mod_codes.GRAY_CODE:

return digital.constellation_qpsk()

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

# DQPSK constellation

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

def dqpsk_constellation(mod_code=_def_mod_code):

if mod_code != mod_codes.GRAY_CODE:

return digital.constellation_dqpsk()

#

# Add these to the mod/demod registry

#

import numpy

def soft_dec_table_generator(soft_dec_gen, prec, Es=1):

'''

| Builds a LUT that is a list of tuples. The tuple represents the

'''

npts = int(2.0**prec)

yrng = numpy.linspace(-maxd, maxd, npts)

xrng = numpy.linspace(-maxd, maxd, npts)

table.append(decs)

return table

def soft_dec_table(constel, symbols, prec, npwr=1):

'''

Similar in nature to soft_dec_table_generator above. Instead, this

table.append(decs)

return table

def calc_soft_dec_from_table(sample, table, prec, Es=1.0):

'''

Takes in a complex sample and converts it from the coordinates

constellation.

'''

lut_scale = 2.0**prec

xre = sample.real

xim = sample.imag

max_index = lut_scale**2

while(index >= max_index):

while(index < 0):

return table[int(index)]

def calc_soft_dec(sample, constel, symbols, npwr=1):

'''

This function takes in any consteallation and symbol symbol set

M = len(constel)

k = int(numpy.log2(M))

for i in range(M):

# Calculate the distance between the sample and the current

for j in range(k):

# Get the bit at the jth index

bit = (symbols[i] & mask) >> j

# If the bit is a 0, add to the probability of a zero

if(bit == 0):

# else, add to the probability of a one

else:

# Calculate the log-likelihood ratio for all bits based on the

# probability of ones (tmp[2*i+1]) over the probability of a zero

# (tmp[2*i+0]).

for i in range(k):

return s

pp = ""

subi = 1

subj = 0

subi = 0

continue

pp += "| "

subj = 1

else:

pp += "( "

pp += "{0: .4f} ".format(item[t])

pp += ") "

pp += "\n"

#

from matplotlib import pyplot

from gnuradio import digital

from .soft_dec_lut_gen import soft_dec_table, calc_soft_dec_from_table, calc_soft_dec

from .psk_constellations import psk_4_0, psk_4_1, psk_4_2, psk_4_3, psk_4_4, psk_4_5, psk_4_6, psk_4_7, sd_psk_4_0, sd_psk_4_1, sd_psk_4_2, sd_psk_4_3, sd_psk_4_4, sd_psk_4_5, sd_psk_4_6, sd_psk_4_7

from .qam_constellations import qam_16_0, sd_qam_16_0

def test_qpsk(i, sample, prec):

qpsk_const_list = [psk_4_0, psk_4_1, psk_4_2, psk_4_3,

psk_4_4, psk_4_5, psk_4_6, psk_4_7]

# Get max energy/symbol in constellation

constel = c.points()

#table = soft_dec_table_generator(qpsk_lut_gen, prec, Es)

table = soft_dec_table(constel, code, prec)

return (y_python_gen_calc, y_python_table, y_python_raw_calc,

y_cpp_table, y_cpp_raw_calc, constel, code, c)

def test_qam16(i, sample, prec):

sample = sample / 1

qam_const_list = [qam_16_0, ]

#table = soft_dec_table_generator(qam_lut_gen, prec, Es)

table = soft_dec_table(constel, code, prec, 1)

c.set_soft_dec_lut(table, prec)

y_python_gen_calc = qam_lut_gen(sample, Es)

return (y_python_gen_calc, y_python_table, y_python_raw_calc,

y_cpp_table, y_cpp_raw_calc, constel, code, c)

if __name__ == "__main__":

index = 0

prec = 8

#x = -1 + -0.j

if 1:

print("C++ Raw calc: ", (y_cpp_raw_calc))

fig = pyplot.figure(1)

sp1.plot([c.real for c in constel],

[c.imag for c in constel], 'bo')

sp1.plot(x.real, x.imag, 'ro')

sp1.set_xlim([-1.5, 1.5])

sp1.set_ylim([-1.5, 1.5])

fill = int(numpy.log2(len(constel)))

ha='center', va='center', size=18)

pyplot.show()

#!/usr/bin/env python

#

# Copyright 2011 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

from .lfsr import lfsr_args

#!/usr/bin/env python

#

# Copyright 2011 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

"""

This module contains functions for aligning sequences.

# The maximum number of samples to take from two sequences to check alignment.

def_num_samples = 1000

def compare_sequences(d1, d2, offset, sample_indices=None):

"""

Takes two binary sequences and an offset and returns the number of

offset -- offset of d2 relative to d1

sample_indices -- a list of indices to use for the comparison

"""

if sample_indices is None:

sample_indices = list(range(0, max_index))

correct = 0

for i in sample_indices:

if i >= max_index:

break

correct += 1

total += 1

return (correct, total)

def random_sample(size, num_samples=def_num_samples, seed=None):

"""

Returns a set of random integers between 0 and (size-1).

num_samples = num_samples / 2

indices = set([])

while len(indices) < num_samples:

indices.add(index)

indices = list(indices)

indices.sort()

return indices

def align_sequences(d1, d2,

num_samples=def_num_samples,

max_offset=def_max_offset,

int_range = [item for items in zip(pos_range, neg_range) for item in items]

for offset in int_range:

correct, compared = compare_sequences(d1, d2, offset, indices)

if frac_correct > max_frac_correct:

max_frac_correct = frac_correct

best_offset = offset

if frac_correct > correct_cutoff:

break

return max_frac_correct, best_compared, best_offset, indices

if __name__ == "__main__":

import doctest

doctest.testmod()

else:

# if not we take advantage of the symmetry of all but the last bit

# around a power of two.

self.gcs.append(result)

self.i += 1

if self.i == self.np2:

self.lp2 = self.i

_gray_code_generator = GrayCodeGenerator()

gray_code = _gray_code_generator.get_gray_code

#!/usr/bin/env python

#

# Copyright 2020 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

def lfsr_args(seed, *exp):

"""

Creates an lfsr object with seed 0b11001, mask 0b1000011, K=6

"""

from functools import reduce

#!/usr/bin/env python

#

# Copyright 2011 Free Software Foundation, Inc.

# This file is part of GNU Radio

# SPDX-License-Identifier: GPL-3.0-or-later

#

# Constants used to represent what coding to use.

GRAY_CODE = 'gray'

codes = (GRAY_CODE, SET_PARTITION_CODE, NO_CODE)

def invert_code(code):

c = enumerate(code)

ic = [(b, a) for (a, b) in c]

from gnuradio import gr

import pmt

def make_lengthtags(lengths, offsets, tagname='length', vlen=1):

tags = []

assert(len(offsets) == len(lengths))

tags.append(tag)

return tags

def string_to_vector(string):

v = []

for s in string:

v.append(ord(s))

return v

def strings_to_vectors(strings, lengthtagname):

vs = [string_to_vector(string) for string in strings]

return packets_to_vectors(vs, lengthtagname)

def vector_to_string(v):

s = []

for d in v:

s.append(chr(d))

return ''.join(s)

def vectors_to_strings(data, tags, lengthtagname):

packets = vectors_to_packets(data, tags, lengthtagname)

return [vector_to_string(packet) for packet in packets]

def count_bursts(data, tags, lengthtagname, vlen=1):

lengthtags = [t for t in tags

if pmt.symbol_to_string(t.key) == lengthtagname]

raise ValueError(

"More than one tags with key {0} with the same offset={1}."

.format(lengthtagname, tag.offset))

in_burst = False

in_packet = False

packet_length = None

for pos in range(len(data)):

if pos in lengths:

if in_packet:

raise Exception("Received packet tag while in packet.")

packet_pos = -1

packet_length = lengths[pos]

in_burst = False

if in_packet:

packet_pos += 1

in_packet = False

packet_pos = None

return burst_count

def vectors_to_packets(data, tags, lengthtagname, vlen=1):

lengthtags = [t for t in tags

if pmt.symbol_to_string(t.key) == lengthtagname]

raise ValueError(

"More than one tags with key {0} with the same offset={1}."

.format(lengthtagname, tag.offset))

if 0 not in lengths:

raise ValueError("There is no tag with key {0} and an offset of 0"

.format(lengthtagname))

length = lengths[pos]

if length == 0:

raise ValueError("Packets cannot have zero length.")

raise ValueError("The final packet is incomplete.")

pos += length

return packets

def packets_to_vectors(packets, lengthtagname, vlen=1):

tags = []

data = []