# First, generate a regular LDPC parity check matrix. Specify

# the properties desired. For example:

# Richardson and Urbanke's preprocessing method requires a full rank

# matrix to start. The matrices generated by the

# Next, some preprocessing steps need to be performed as described

# Richardson and Urbanke in Modern Coding Theory, Appendix A. This

# can take a while...

# Print(out some of the resulting properties.)

n = bestH.shape[1]

print("Parity check matrix properties:")

print("\tSize :", bestH.shape)

print("\tRank :", linalg.matrix_rank(bestH))

print("\tn :", n, " (codeword length)")

print("\tk :", k, " (info word length)")

print("\tgap : %i" % g)

# Save the matrix to an alist file for future use:

class extended_decoder(gr.hier_block2):

garbletable = {

}

gr.log.info("fec.extended_decoder: Parallelism must be 1.")

raise AttributeError

else:

# If it has parallelism of 0, force it into a list of 1

self.blocks.append(blocks.multiply_const_ff(48.0))

if fec.get_shift(decoder_obj_list[0]) != 0.0:

elif fec.get_decoder_input_conversion(decoder_obj_list[0]) == "packed_bits":

self.blocks.append(blocks.add_const_ff(128.0))

if not self.flush:

else:

for i in fec.read_big_bitlist(ann):

for i in idx_list:

synd_garble = self.garbletable[i]

if fec.get_decoder_input_conversion(decoder_obj_list[0]) == "packed_bits":

self.blocks.append(blocks.uchar_to_float())

self.blocks.append(blocks.add_const_ff(-128.0))

self.blocks.append(digital.binary_slicer_fb())

raise AttributeError

else:

if fec.get_decoder_output_conversion(decoder_obj_list[0]) == "unpack":

for i in range(len(self.blocks) - 1):

class extended_tagged_decoder(gr.hier_block2):

garbletable = {

}

# This block doesn't handle parallelism of > 1

gr.log.info("fec.extended_tagged_decoder: Parallelism must be 1.")

raise AttributeError

# If lentagname is None, fall back to using the non tagged

# stream version

lentagname = None

self.blocks.append(blocks.multiply_const_ff(48.0))

if fec.get_shift(decoder_obj) != 0.0:

elif fec.get_decoder_input_conversion(decoder_obj) == "packed_bits":

self.blocks.append(blocks.add_const_ff(128.0))

if not self.flush:

else:

for i in fec.read_big_bitlist(ann):

for i in idx_list:

synd_garble = self.garbletable[i]

if fec.get_decoder_input_conversion(decoder_obj) == "packed_bits":

self.blocks.append(blocks.uchar_to_float())

self.blocks.append(blocks.add_const_ff(-128.0))

self.blocks.append(digital.binary_slicer_fb())

else:

else:

if fec.get_decoder_output_conversion(decoder_obj) == "unpack":

for i in range(len(self.blocks) - 1):

#

#

[0] Erdal Arikan: 'Channel Polarization: A Method for Constructing Capacity-Achieving Codes for Symmetric Binary-Input Memoryless Channels', 2009

foundational paper for polar codes.

from .channel_construction_bec import calculate_bec_channel_capacities

from .channel_construction_bec import design_snr_to_bec_eta

from .channel_construction_bec import bhattacharyya_bounds

import numpy as np

try:

from .channel_construction_awgn import tal_vardy_tpm_algorithm

except ImportError:

print("SciPy missing. Overwrite Tal-Vardy algorithm with BEC approximation")

def tal_vardy_tpm_algorithm(block_size, design_snr, mu):

return bhattacharyya_bounds(design_snr, block_size)

def frozen_bit_positions(block_size, info_size, design_snr=0.0):

if not design_snr > -1.5917:

design_snr = 0.0

eta = design_snr_to_bec_eta(design_snr)

return get_bec_frozen_indices(block_size, block_size - info_size, eta)

def default_dir():

dir_def = "~/.gnuradio/polar/"

import os

path = os.path.expanduser(dir_def)

try:

return path

path = default_dir()

filename = generate_filename(block_size, design_snr, mu)

header = Z_PARAM_FIRST_HEADER_LINE + "\n"

filename = generate_filename(block_size, design_snr, mu)

full_file = path + filename

import os

if not os.path.isfile(full_file):

z_params = tal_vardy_tpm_algorithm(block_size, design_snr, mu)

save_z_parameters(z_params, block_size, design_snr, mu)

def main():

np.set_printoptions(precision=3, linewidth=150)

n = 10

m = 2 ** n

k = m // 2

if 0:

import matplotlib.pyplot as plt

plt.plot(z_params)

plt.plot(z_bounds)

plt.show()

main()

#

Based on 2 papers:

[1] Ido Tal, Alexander Vardy: 'How To Construct Polar Codes', 2013

for an in-depth description of a widely used algorithm for channel construction.

[2] Harish Vangala, Emanuele Viterbo, Yi Hong: 'A Comparative Study of Polar Code Constructions for the AWGN Channel', 2015

for an overview of different approaches

from scipy.optimize import fsolve

from scipy.special import erfc

import numpy as np

from .channel_construction_bec import bhattacharyya_bounds

def instantanious_capacity_callable():

def instantanious_capacity(x):

def q_function(x):

# Q(x) = (1 / sqrt(2 * pi) ) * integral (x to inf) exp(- x ^ 2 / 2) dx

def discretize_awgn(mu, design_snr):

needed for Binary-AWGN channels.

in [1] described in Section VI

in [2] described as a function of the same name.

2. split into mu intervals.

3. find corresponding output alphabet values y of likelihood ratio function lambda(y) inserted into C(x)

4. Calculate probability for each value given that a '0' or '1' is was transmitted.

s = 10 ** (design_snr / 10)

a = np.zeros(mu + 1, dtype=float)

a[-1] = np.inf

for i in range(1, mu):

factor = np.sqrt(2 * s)

tpm = np.zeros((2, mu))

for j in range(mu):

tpm[0][j] = q_function(factor + a[j]) - q_function(factor + a[j + 1])

tpm = tpm[::-1]

tpm[0] = tpm[0][::-1]

def instant_capacity_delta_callable():

def capacity_delta_callable():

calculate_delta_I = capacity_delta_callable()

L = np.shape(tpm)[1]

if not mu < L:

# lambda works on vectors just fine. Use Numpy vector awesomeness.

delta_i_vec = calculate_delta_I(tpm[0, 0:-1], tpm[1, 0:-1], tpm[0, 1:], tpm[1, 1:])

channels = np.zeros((block_size, 2, mu))

channels[0] = discretize_awgn(mu, design_snr) * 2

show_progress_bar(0, block_size)

for j in range(0, block_power):

u = 2 ** j

z = z[bit_reverse_vector(np.arange(block_size), block_power)]

z = upper_bound_z_params(z, block_size, design_snr)

show_progress_bar(block_size, block_size)

return z

def main():

n = 8

m = 2 ** n

design_snr = 0.0

if 0:

import matplotlib.pyplot as plt

plt.plot(z_params)

plt.show()

main()

#

import numpy as np

from .helper_functions import bit_reverse_vector, is_power_of_two

PolarCommon holds value checks and common initializer code for both Encoder and Decoder.

class PolarCommon(object):

if frozenbits is None:

frozenbits = np.zeros(n - k, dtype=np.int)

if not len(frozenbits) == n - k:

if not frozenbits.dtype == np.int:

frozenbits = frozenbits.astype(dtype=int)

if not len(frozen_bit_position) == (n - k):

if not frozen_bit_position.dtype == np.int:

frozen_bit_position = frozen_bit_position.astype(dtype=int)

self.N = n

self.power = int(np.log2(self.N))

self.K = k

return vec

def _encode_natural_order(self, vec):

vec = vec[self.bit_reverse_positions]

return self._encode_efficient(vec)

#

from .encoder import PolarEncoder

from .decoder import PolarDecoder

from . import channel_construction as cc

# frozenbitposition = np.array((0, 1, 2, 3, 4, 5, 8, 9), dtype=int)

m = 256

n_frozen = m // 2

print(frozenbitposition)

return frozenbitposition

encoded = encoder.encode(bits)

rx = decoder.decode(encoded)

if not is_equal(bits, rx):

return

if not (first == second).all():

result = first == second

for i in range(len(result)):

return False

return True

def path_metric_exact(last_pm, llr, ui):

def path_metric_approx(last_pm, llr, ui):

return last_pm

return last_pm + np.abs(llr)

bits = np.random.randint(2, size=k)

tx = encoder.encode(bits)

np.random.shuffle(possible_indices)

rx = tx

recv = decoder.decode(rx)

print(channel_counter)

print(np.min(channel_counter), np.max(channel_counter))

def find_good_indices(res, nindices):

def channel_analysis():

ntests = 10000

channel_counter = np.load(filename)

print(np.min(channel_counter), np.max(channel_counter))

channel_counter[0] = np.min(channel_counter)

good_indices = find_good_indices(channel_counter, channel_counter.size // 2)

info_bit_positions = np.where(good_indices > 0)

print(info_bit_positions)

print(frozen_bit_positions)

good_indices *= 2000

good_indices += 4000

def merge_first_stage(init_mask):

merged_frozen_mask = []

for e in range(0, len(init_mask), 2):

else:

return merged_frozen_mask

def merge_second_stage(init_mask):

merged_frozen_mask = []

for e in range(0, len(init_mask), 2):

else:

merged_frozen_mask.append(init_mask[e])

merged_frozen_mask.append(init_mask[e + 1])

merged_frozen_mask = []

n_elems = len(init_mask) - (len(init_mask) % 2)

for e in range(0, n_elems, 2):

else:

merged_frozen_mask.append(init_mask[e])

merged_frozen_mask.append(init_mask[e + 1])

sub_mask = mask.flatten()

lock_mask = lock.flatten()

ll = lock_mask[0]

sub_t = sub_mask[0]

for i in range(len(frozen_mask)):

# if i % 8 == 0:

# print

if not l == ll or not sub_mask[i] == sub_t:

ll = l

sub_t = sub_mask[i]

def systematic_encoder_decoder_chain_test():

block_size = int(2 ** 8)

info_bit_size = block_size // 2

ntests = 100

encoder = PolarEncoder(block_size, info_bit_size, frozenbitposition)

decoder = PolarDecoder(block_size, info_bit_size, frozenbitposition)

for i in range(ntests):

systematic_encoder_decoder_chain_test()

main()

draw,

figtext,

figure,

get_current_fig_manager,

rcParams,

searchsorted,

from matplotlib.font_manager import fontManager, FontProperties

except ImportError:

raise SystemExit(1)

from argparse import ArgumentParser

class draw_constellation:

def __init__(self, filename, options):

self.hfile = open(filename, "r")

self.sample_rate = options.sample_rate

self.datatype = numpy.complex64

self.axis_font_size = 16

self.label_font_size = 18

self.title_font_size = 20

# Setup PLOT

self.make_plots()

self.button_left = Button(self.button_left_axes, "<")

self.button_left_callback = self.button_left.on_clicked(self.button_left_click)

self.button_right = Button(self.button_right_axes, ">")

self.xlim = self.sp_iq.get_xlim()

self.manager = get_current_fig_manager()

show()

def get_data(self):

try:

except MemoryError:

print("End of File")

else:

# retesting length here as newer version of numpy does not throw a MemoryError, just

# returns a zero-length array

self.reals = numpy.array([r.real for r in iq])

self.imags = numpy.array([i.imag for i in iq])

return True

else:

print("End of File")

def make_plots(self):

# if specified on the command-line, set file pointer

r = self.get_data()

# Subplot for real and imaginary parts of signal

self.sp_iq.set_title(("I&Q"), fontsize=self.title_font_size, fontweight="bold")

# Subplot for constellation plot

self.indx = 0

# Adjust axis

self.sp_const.axis([-2, 2, -2, 2])

draw()

def update_plots(self):

self.plot_iq[0].set_data([self.time, self.reals])

self.plot_iq[1].set_data([self.time, self.imags])

self.plot_const[0].set_data([self.reals, self.imags])

self.sp_const.axis([-2, 2, -2, 2])

def zoom(self, event):

newxlim = numpy.array(self.sp_iq.get_xlim())

curxlim = numpy.array(self.xlim)

self.xlim = newxlim

r = self.reals[int(ceil(self.xlim[0])) : int(ceil(self.xlim[1]))]

i = self.imags[int(ceil(self.xlim[0])) : int(ceil(self.xlim[1]))]

def click(self, event):

forward_valid_keys = [" ", "down", "right"]

backward_valid_keys = ["up", "left"]

self.step_forward()

self.step_backward()

self.set_trace(self.indx)

self.set_trace(self.indx)

def button_left_click(self, event):

def step_forward(self):

r = self.get_data()

self.update_plots()

def step_backward(self):

# Step back in file position

else:

r = self.get_data()

self.update_plots()

def mouse_button_callback(self, event):

x, y = event.xdata, event.ydata

if x is not None and y is not None:

self.indx = searchsorted(self.time, [x])

self.set_trace(self.indx)

def set_trace(self, indx):

self.plot_iq[2].set_data(self.time[indx], self.reals[indx])

self.plot_iq[3].set_data(self.time[indx], self.imags[indx])

def find(item_in, list_search):

try:

except ValueError:

return False

description = "Takes a GNU Radio complex binary file and displays the I&Q data versus time and the constellation plot (I vs. Q). You can set the block size to specify how many points to read in at a time and the start position in the file. By default, the system assumes a sample rate of 1, so in time, each sample is plotted versus the sample number. To set a true time axis, set the sample rate (-R or --sample-rate) to the sample rate used when capturing the samples."

parser = ArgumentParser(conflict_handler="resolve", description=description)

args = parser.parse_args()

dc = draw_constellation(args.file, args)

if __name__ == "__main__":

try:

main()

import numpy

try:

except ImportError:

raise SystemExit(1)

from argparse import ArgumentParser

class draw_iq:

def __init__(self, filename, options):

self.hfile = open(filename, "r")

self.sample_rate = options.sample_rate

self.datatype = numpy.complex64

self.axis_font_size = 16

self.label_font_size = 18

self.text_size = 22

# Setup PLOT

self.make_plots()

self.button_left = Button(self.button_left_axes, "<")

self.button_left_callback = self.button_left.on_clicked(self.button_left_click)

self.button_right = Button(self.button_right_axes, ">")

self.xlim = self.sp_iq.get_xlim()

self.manager = get_current_fig_manager()

show()

def get_data(self):

try:

except MemoryError:

print("End of File")

else:

self.reals = numpy.array([r.real for r in self.iq])

self.imags = numpy.array([i.imag for i in self.iq])

def make_plots(self):

# if specified on the command-line, set file pointer

self.get_data()

# Subplot for real and imaginary parts of signal

self.sp_iq.set_title(("I&Q"), fontsize=self.title_font_size, fontweight="bold")

self.sp_iq.set_xlim(self.time.min(), self.time.max())

draw()

def update_plots(self):

self.plot_iq[0].set_data([self.time, self.reals])

self.plot_iq[1].set_data([self.time, self.imags])

self.sp_iq.set_xlim(self.time.min(), self.time.max())

draw()

forward_valid_keys = [" ", "down", "right"]

backward_valid_keys = ["up", "left"]

self.step_forward()

self.step_backward()

def button_left_click(self, event):

def step_backward(self):

# Step back in file position

else:

self.get_data()

self.update_plots()

def find(item_in, list_search):

try:

except ValueError:

return False

def main():

description = "Takes a GNU Radio complex binary file and displays the I&Q data versus time. You can set the block size to specify how many points to read in at a time and the start position in the file. By default, the system assumes a sample rate of 1, so in time, each sample is plotted versus the sample number. To set a true time axis, set the sample rate (-R or --sample-rate) to the sample rate used when capturing the samples."

parser = ArgumentParser(conflict_handler="resolve", description=description)

args = parser.parse_args()

dc = draw_iq(args.file, args)

if __name__ == "__main__":

try:

main()