summaryrefslogtreecommitdiff
path: root/gr-digital/python
diff options
context:
space:
mode:
authorBen Reynwar <ben@reynwar.net>2013-05-10 05:45:12 -0700
committerBen Reynwar <ben@reynwar.net>2013-05-10 05:45:12 -0700
commite4f0319eced22c112f7e6a4cc45bc2036d285332 (patch)
tree92baa4254f47fd0b62d1d3f69e3b1ddb96dd32d3 /gr-digital/python
parent5fd6b62702369e45ef01a02e12bec1afc5b57200 (diff)
parent3e052d303874a07237c4e0c31eef3f7f7a192835 (diff)
Merged in 'next' branch.
Diffstat (limited to 'gr-digital/python')
-rwxr-xr-xgr-digital/python/digital/qa_mpsk_snr_est.py2
-rwxr-xr-xgr-digital/python/digital/qa_ofdm_frame_equalizer_vcvc.py33
-rwxr-xr-xgr-digital/python/digital/qa_ofdm_serializer_vcc.py3
-rwxr-xr-xgr-digital/python/digital/qa_ofdm_sync_sc_cfb.py3
-rwxr-xr-xgr-digital/python/digital/qa_packet_headergenerator_bb.py8
-rwxr-xr-xgr-digital/python/digital/qa_packet_headerparser_b.py97
6 files changed, 119 insertions, 27 deletions
diff --git a/gr-digital/python/digital/qa_mpsk_snr_est.py b/gr-digital/python/digital/qa_mpsk_snr_est.py
index cb3e954141..032edf1c73 100755
--- a/gr-digital/python/digital/qa_mpsk_snr_est.py
+++ b/gr-digital/python/digital/qa_mpsk_snr_est.py
@@ -87,7 +87,7 @@ class test_mpsk_snr_est(gr_unittest.TestCase):
self.assertFloatTuplesAlmostEqual(expected_result, actual_result, 2)
def test_mpsk_snr_est_svn(self):
- expected_result = [10.90, 6.00, 4.76, 4.97, 5.49]
+ expected_result = [10.92, 6.02, 4.78, 4.98, 5.51]
N = 10000
alpha = 0.001
diff --git a/gr-digital/python/digital/qa_ofdm_frame_equalizer_vcvc.py b/gr-digital/python/digital/qa_ofdm_frame_equalizer_vcvc.py
index ee1e849d92..3b3b896b83 100755
--- a/gr-digital/python/digital/qa_ofdm_frame_equalizer_vcvc.py
+++ b/gr-digital/python/digital/qa_ofdm_frame_equalizer_vcvc.py
@@ -105,6 +105,39 @@ class qa_ofdm_frame_equalizer_vcvc (gr_unittest.TestCase):
if pmt.symbol_to_string(tag.key) == "ofdm_sync_chan_taps":
self.assertEqual(list(pmt.c32vector_elements(tag.value)), channel[-fft_len:])
+ def test_002_static_wo_tags (self):
+ fft_len = 8
+ # 4 5 6 7 0 1 2 3
+ tx_data = [-1, -1, 1, 2, -1, 3, 0, -1, # 0
+ -1, -1, 0, 2, -1, 2, 0, -1, # 8
+ -1, -1, 3, 0, -1, 1, 0, -1, # 16 (Pilot symbols)
+ -1, -1, 1, 1, -1, 0, 2, -1] # 24
+ cnst = digital.constellation_qpsk()
+ tx_signal = [cnst.map_to_points_v(x)[0] if x != -1 else 0 for x in tx_data]
+ occupied_carriers = ((1, 2, 6, 7),)
+ pilot_carriers = ((), (), (1, 2, 6, 7), ())
+ pilot_symbols = (
+ [], [], [cnst.map_to_points_v(x)[0] for x in (1, 0, 3, 0)], []
+ )
+ equalizer = digital.ofdm_equalizer_static(fft_len, occupied_carriers, pilot_carriers, pilot_symbols)
+ channel = [
+ 0, 0, 1, 1, 0, 1, 1, 0,
+ 0, 0, 1, 1, 0, 1, 1, 0, # These coefficients will be rotated slightly...
+ 0, 0, 1j, 1j, 0, 1j, 1j, 0, # Go crazy here!
+ 0, 0, 1j, 1j, 0, 1j, 1j, 0 # ...and again here.
+ ]
+ for idx in range(fft_len, 2*fft_len):
+ channel[idx] = channel[idx-fft_len] * numpy.exp(1j * .1 * numpy.pi * (numpy.random.rand()-.5))
+ idx2 = idx+2*fft_len
+ channel[idx2] = channel[idx2] * numpy.exp(1j * 0 * numpy.pi * (numpy.random.rand()-.5))
+ src = blocks.vector_source_c(numpy.multiply(tx_signal, channel), False, fft_len)
+ eq = digital.ofdm_frame_equalizer_vcvc(equalizer.base(), "", False, 4)
+ sink = blocks.vector_sink_c(fft_len)
+ self.tb.connect(src, eq, sink)
+ self.tb.run ()
+ rx_data = [cnst.decision_maker_v((x,)) if x != 0 else -1 for x in sink.data()]
+ self.assertEqual(tx_data, rx_data)
+
def test_002_simpledfe (self):
fft_len = 8
# 4 5 6 7 0 1 2 3
diff --git a/gr-digital/python/digital/qa_ofdm_serializer_vcc.py b/gr-digital/python/digital/qa_ofdm_serializer_vcc.py
index fb1f47f2fd..8a5e8f4893 100755
--- a/gr-digital/python/digital/qa_ofdm_serializer_vcc.py
+++ b/gr-digital/python/digital/qa_ofdm_serializer_vcc.py
@@ -207,6 +207,5 @@ class qa_ofdm_serializer_vcc (gr_unittest.TestCase):
if __name__ == '__main__':
- #gr_unittest.run(qa_ofdm_serializer_vcc, "qa_ofdm_serializer_vcc.xml")
- gr_unittest.run(qa_ofdm_serializer_vcc)
+ gr_unittest.run(qa_ofdm_serializer_vcc, "qa_ofdm_serializer_vcc.xml")
diff --git a/gr-digital/python/digital/qa_ofdm_sync_sc_cfb.py b/gr-digital/python/digital/qa_ofdm_sync_sc_cfb.py
index ba50240916..3e811767cd 100755
--- a/gr-digital/python/digital/qa_ofdm_sync_sc_cfb.py
+++ b/gr-digital/python/digital/qa_ofdm_sync_sc_cfb.py
@@ -194,6 +194,5 @@ Detection error was: %d """ % (numpy.sum(sink_detect.data()) - n_bursts)
if __name__ == '__main__':
- #gr_unittest.run(qa_ofdm_sync_sc_cfb, "qa_ofdm_sync_sc_cfb.xml")
- gr_unittest.run(qa_ofdm_sync_sc_cfb)
+ gr_unittest.run(qa_ofdm_sync_sc_cfb, "qa_ofdm_sync_sc_cfb.xml")
diff --git a/gr-digital/python/digital/qa_packet_headergenerator_bb.py b/gr-digital/python/digital/qa_packet_headergenerator_bb.py
index aaf7f89918..df4512b2a0 100755
--- a/gr-digital/python/digital/qa_packet_headergenerator_bb.py
+++ b/gr-digital/python/digital/qa_packet_headergenerator_bb.py
@@ -31,8 +31,8 @@ class qa_packet_headergenerator_bb (gr_unittest.TestCase):
self.tb = None
def test_001_12bits (self):
- # 3 PDUs: | | | |
- data = (1, 2, 3, 4, 1, 2, 1, 2, 3, 4)
+ # 3 PDUs: | | |
+ data = (1, 2, 3, 4, 1, 2) + tuple(range(25))
tagname = "packet_len"
tag1 = gr.gr_tag_t()
tag1.offset = 0
@@ -45,7 +45,7 @@ class qa_packet_headergenerator_bb (gr_unittest.TestCase):
tag3 = gr.gr_tag_t()
tag3.offset = 6
tag3.key = pmt.string_to_symbol(tagname)
- tag3.value = pmt.from_long(4)
+ tag3.value = pmt.from_long(25)
src = blocks.vector_source_b(data, False, 1, (tag1, tag2, tag3))
header = digital.packet_headergenerator_bb(12, tagname)
sink = blocks.vector_sink_b()
@@ -54,7 +54,7 @@ class qa_packet_headergenerator_bb (gr_unittest.TestCase):
expected_data = (
0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
- 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0
+ 1, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0
)
self.assertEqual(sink.data(), expected_data)
diff --git a/gr-digital/python/digital/qa_packet_headerparser_b.py b/gr-digital/python/digital/qa_packet_headerparser_b.py
index b127c86fb1..8f63ac04d4 100755
--- a/gr-digital/python/digital/qa_packet_headerparser_b.py
+++ b/gr-digital/python/digital/qa_packet_headerparser_b.py
@@ -20,7 +20,10 @@
#
import time
+import random
+
from gnuradio import gr, gr_unittest, blocks, digital
+from gnuradio.digital import tagged_streams
import pmt
class qa_packet_headerparser_b (gr_unittest.TestCase):
@@ -32,38 +35,96 @@ class qa_packet_headerparser_b (gr_unittest.TestCase):
self.tb = None
def test_001_t (self):
- expected_data = (
- # | Number of symbols | Packet number | Parity
+ """
+ First header: Packet length 4, packet num 0
+ Second header: Packet 2, packet num 1
+ Third header: Invalid (parity bit does not check) (would be len 4, num 2)
+ """
+ encoded_headers = (
+ # | Number of bytes | Packet number | Parity
0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0,
0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0
)
- tagname = "packet_len"
+ packet_len_tagname = "packet_len"
- src = blocks.vector_source_b(expected_data)
- parser = digital.packet_headerparser_b(32, tagname)
+ src = blocks.vector_source_b(encoded_headers)
+ parser = digital.packet_headerparser_b(32, packet_len_tagname)
sink = blocks.message_debug()
-
self.tb.connect(src, parser)
self.tb.msg_connect(parser, "header_data", sink, "store")
- self.tb.start ()
+ self.tb.start()
time.sleep(1)
self.tb.stop()
self.tb.wait()
-
self.assertEqual(sink.num_messages(), 3)
- msg = sink.get_message(0)
- #try:
- #self.assertEqual(4, pmt.pmt_to_long(pmt.pmt_dict_ref(msg, pmt.pmt_string_to_symbol(tagname), pmt.PMT_F)))
- #self.assertEqual(0, pmt.pmt_to_long(pmt.pmt_dict_ref(msg, pmt.pmt_string_to_symbol("packet_num"), pmt.PMT_F)))
-
- #except:
- #self.fail()
- # msg1: length 4, number 0
- # msg2: length 2, number 1
- # msg3: PMT_F because parity fail
+ msg1 = pmt.to_python(sink.get_message(0))
+ msg2 = pmt.to_python(sink.get_message(1))
+ msg3 = pmt.to_python(sink.get_message(2))
+ self.assertEqual(msg1, {'packet_len': 4, 'packet_num': 0})
+ self.assertEqual(msg2, {'packet_len': 2, 'packet_num': 1})
+ self.assertEqual(msg3, False)
+ def test_002_pipe(self):
+ """
+ Create N packets of random length, pipe them through header generator,
+ back to header parser, make sure output is the same.
+ """
+ N = 20
+ header_len = 32
+ packet_len_tagname = "packet_len"
+ packet_lengths = [random.randint(1, 100) for x in range(N)]
+ data, tags = tagged_streams.packets_to_vectors([range(packet_lengths[i]) for i in range(N)], packet_len_tagname)
+ src = blocks.vector_source_b(data, False, 1, tags)
+ header_gen = digital.packet_headergenerator_bb(header_len, packet_len_tagname)
+ header_parser = digital.packet_headerparser_b(header_len, packet_len_tagname)
+ sink = blocks.message_debug()
+ self.tb.connect(src, header_gen, header_parser)
+ self.tb.msg_connect(header_parser, "header_data", sink, "store")
+ self.tb.start()
+ time.sleep(1)
+ self.tb.stop()
+ self.tb.wait()
+ self.assertEqual(sink.num_messages(), N)
+ for i in xrange(N):
+ msg = pmt.to_python(sink.get_message(i))
+ self.assertEqual(msg, {'packet_len': packet_lengths[i], 'packet_num': i})
+ def test_003_ofdm (self):
+ """ Header 1: 193 bytes
+ Header 2: 8 bytes
+ 2 bits per complex symbol, 32 carriers => 64 bits = 8 bytes per OFDM symbol
+ """
+ encoded_headers = (
+ # | Number of bytes | Packet number | Parity
+ 1, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0,
+ 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+ )
+ packet_len_tagname = "packet_len"
+ frame_len_tagname = "frame_len"
+ src = blocks.vector_source_b(encoded_headers)
+ header_formatter = digital.packet_header_ofdm(
+ (range(32),), # 32 carriers are occupied (which doesn't matter here)
+ 1, # 1 OFDM symbol per header (= 32 bits)
+ packet_len_tagname,
+ frame_len_tagname,
+ "packet_num",
+ 1, # 1 bit per header symbols (BPSK)
+ 2 # 2 bits per payload symbol (QPSK)
+ )
+ parser = digital.packet_headerparser_b(header_formatter.base())
+ sink = blocks.message_debug()
+ self.tb.connect(src, parser)
+ self.tb.msg_connect(parser, "header_data", sink, "store")
+ self.tb.start()
+ time.sleep(1)
+ self.tb.stop()
+ self.tb.wait()
+ self.assertEqual(sink.num_messages(), 2)
+ msg1 = pmt.to_python(sink.get_message(0))
+ msg2 = pmt.to_python(sink.get_message(1))
+ self.assertEqual(msg1, {'packet_len': 193, 'frame_len': 25, 'packet_num': 0})
+ self.assertEqual(msg2, {'packet_len': 8, 'frame_len': 1, 'packet_num': 1})
if __name__ == '__main__':
gr_unittest.run(qa_packet_headerparser_b, "qa_packet_headerparser_b.xml")
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-28 06:12:34 +0000