__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 = []