#!/usr/bin/env python
#
# Copyright 2008,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, eng_notation
from optparse import OptionParser
from gnuradio.eng_option import eng_option
import threading
import sys
import time
import math
from gnuradio import digital
from gnuradio import blocks
# from current dir
from uhd_interface import uhd_receiver
n2s = eng_notation.num_to_str
class status_thread(threading.Thread):
def __init__(self, tb):
threading.Thread.__init__(self)
self.setDaemon(1)
self.tb = tb
self.done = False
self.start()
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(
tb.frequency_offset(), tb.timing_offset() * 1e6, tb.snr(), tb.ber()))
try:
time.sleep(1.0)
except KeyboardInterrupt:
self.done = True
class bert_receiver(gr.hier_block2):
def __init__(self, bitrate,
constellation, samples_per_symbol,
differential, excess_bw, gray_coded,
freq_bw, timing_bw, phase_bw,
verbose, log):
gr.hier_block2.__init__(self, "bert_receive",
# Input signature
gr.io_signature(1, 1, gr.sizeof_gr_complex),
gr.io_signature(0, 0, 0)) # Output signature
self._bitrate = bitrate
self._demod = digital.generic_demod(constellation, differential,
samples_per_symbol,
gray_coded, excess_bw,
freq_bw, timing_bw, phase_bw,
verbose, log)
self._symbol_rate = self._bitrate / self._demod.bits_per_symbol()
self._sample_rate = self._symbol_rate * samples_per_symbol
# Add an SNR probe on the demodulated constellation
self._snr_probe = digital.probe_mpsk_snr_est_c(digital.SNR_EST_M2M4, 1000,
alpha=10.0 / self._symbol_rate)
self.connect(self._demod.time_recov, self._snr_probe)
# Descramble BERT sequence. A channel error will create 3 incorrect bits
self._descrambler = digital.descrambler_bb(
0x8A, 0x7F, 7) # CCSDS 7-bit descrambler
# 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):
return self._demod.freq_recov.get_frequency() * self._sample_rate / (2 * math.pi)
def timing_offset(self):
return self._demod.time_recov.clock_rate()
def snr(self):
return self._snr_probe.snr()
def ber(self):
return (1.0 - self._ber.density()) / 3.0
class rx_psk_block(gr.top_block):
def __init__(self, demod, options):
gr.top_block.__init__(self, "rx_mpsk")
self._demodulator_class = demod
# Get demod_kwargs
demod_kwargs = self._demodulator_class.extract_kwargs_from_options(
options)
# demodulator
self._demodulator = self._demodulator_class(**demod_kwargs)
if(options.rx_freq is not None):
symbol_rate = options.bitrate / self._demodulator.bits_per_symbol()
self._source = uhd_receiver(options.args, symbol_rate,
options.samples_per_symbol,
options.rx_freq, options.rx_gain,
options.spec,
options.antenna, options.verbose)
options.samples_per_symbol = self._source._sps
elif(options.from_file is not None):
self._source = blocks.file_source(
gr.sizeof_gr_complex, options.from_file)
else:
self._source = blocks.null_source(gr.sizeof_gr_complex)
# Create the BERT receiver
self._receiver = bert_receiver(options.bitrate,
self._demodulator._constellation,
options.samples_per_symbol,
options.differential,
options.excess_bw,
gray_coded=True,
freq_bw=options.freq_bw,
timing_bw=options.timing_bw,
phase_bw=options.phase_bw,
verbose=options.verbose,
log=options.log)
self.connect(self._source, self._receiver)
def snr(self):
return self._receiver.snr()
def mag(self):
return self._receiver.signal_mean()
def var(self):
return self._receiver.noise_variance()
def ber(self):
return self._receiver.ber()
def frequency_offset(self):
return self._receiver.frequency_offset()
def timing_offset(self):
return self._receiver.timing_offset()
def get_options(demods):
parser = OptionParser(option_class=eng_option, conflict_handler="resolve")
parser.add_option("", "--from-file", default=None,
help="input file of samples to demod")
parser.add_option("-m", "--modulation", type="choice", choices=list(demods.keys()),
default='psk',
help="Select modulation from: %s [default=%%default]"
% (', '.join(list(demods.keys())),))
parser.add_option("-r", "--bitrate", type="eng_float", default=250e3,
help="Select modulation bit rate (default=%default)")
parser.add_option("-S", "--samples-per-symbol", type="float", default=2,
help="set samples/symbol [default=%default]")
if not parser.has_option("--verbose"):
parser.add_option("-v", "--verbose",
action="store_true", default=False)
if not parser.has_option("--log"):
parser.add_option("", "--log", action="store_true", default=False,
help="Log all parts of flow graph to files (CAUTION: lots of data)")
uhd_receiver.add_options(parser)
demods = digital.modulation_utils.type_1_demods()
for mod in list(demods.values()):
mod.add_options(parser)
(options, args) = parser.parse_args()
if len(args) != 0:
parser.print_help()
sys.exit(1)
return (options, args)
if __name__ == "__main__":
print("""Warning: this example in its current shape is deprecated and
will be removed or fundamentally reworked in a coming GNU Radio
release.""")
demods = digital.modulation_utils.type_1_demods()
(options, args) = get_options(demods)
demod = demods[options.modulation]
tb = rx_psk_block(demod, options)
print("\n*** SNR estimator is inaccurate below about 7dB")
print("*** BER estimator is inaccurate above about 10%\n")
updater = status_thread(tb)
try:
tb.run()
except KeyboardInterrupt:
updater.done = True
updater = None