1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
#!/usr/bin/env python
#
# Copyright 2008,2010,2012,2013 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
#
from gnuradio import gr, gr_unittest, digital, blocks
import pmt
# See gr-digital/lib/additive_scrambler_bb_impl.cc for reference.
def additive_scramble_lfsr(mask, seed, reglen, bpb, data):
l = digital.lfsr(mask, seed, reglen)
out = []
for d in data:
scramble_word = 0
for i in range(0,bpb):
scramble_word ^= l.next_bit() << i
out.append(d ^ scramble_word)
return out
class test_scrambler(gr_unittest.TestCase):
def setUp(self):
self.tb = gr.top_block()
def tearDown(self):
self.tb = None
def test_scrambler_descrambler(self):
src_data = [1,]*1000
src = blocks.vector_source_b(src_data, False)
scrambler = digital.scrambler_bb(0x8a, 0x7F, 7) # CCSDS 7-bit scrambler
descrambler = digital.descrambler_bb(0x8a, 0x7F, 7)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, descrambler, dst)
self.tb.run()
self.assertEqual(src_data[:-8], dst.data()[8:]) # skip garbage during synchronization
def test_additive_scrambler(self):
src_data = [1,]*1000
src = blocks.vector_source_b(src_data, False)
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7)
descrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, descrambler, dst)
self.tb.run()
self.assertEqual(src_data, dst.data())
def test_additive_scrambler_reset(self):
src_data = [1,]*200
src = blocks.vector_source_b(src_data, False)
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 50)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, dst)
self.tb.run()
output = dst.data()
self.assertEqual(output[:50] * 4, output)
def test_additive_scrambler_reset_3bpb(self):
src_data = [5,]*200
src = blocks.vector_source_b(src_data, False)
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 50, 3)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, dst)
self.tb.run()
output = dst.data()
self.assertEqual(output[:50] * 4, output)
def test_additive_scrambler_tags(self):
src_data = [1,]*1000
src = blocks.vector_source_b(src_data, False)
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 100)
descrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 100)
reset_tag_key = 'reset_lfsr'
reset_tag1 = gr.tag_t()
reset_tag1.key = pmt.string_to_symbol(reset_tag_key)
reset_tag1.offset = 17
reset_tag2 = gr.tag_t()
reset_tag2.key = pmt.string_to_symbol(reset_tag_key)
reset_tag2.offset = 110
reset_tag3 = gr.tag_t()
reset_tag3.key = pmt.string_to_symbol(reset_tag_key)
reset_tag3.offset = 523
src = blocks.vector_source_b(src_data, False, 1, (reset_tag1, reset_tag2, reset_tag3))
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 100, 1, reset_tag_key)
descrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 100, 1, reset_tag_key)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, descrambler, dst)
self.tb.run()
self.assertEqual(src_data, dst.data())
def test_additive_scrambler_tags_oneway(self):
src_data = [x for x in range(0, 10)]
reset_tag_key = 'reset_lfsr'
reset_tag1 = gr.tag_t()
reset_tag1.key = pmt.string_to_symbol(reset_tag_key)
reset_tag1.offset = 0
reset_tag2 = gr.tag_t()
reset_tag2.key = pmt.string_to_symbol(reset_tag_key)
reset_tag2.offset = 10
reset_tag3 = gr.tag_t()
reset_tag3.key = pmt.string_to_symbol(reset_tag_key)
reset_tag3.offset = 20
src = blocks.vector_source_b(src_data * 3, False, 1, (reset_tag1, reset_tag2, reset_tag3))
scrambler = digital.additive_scrambler_bb(0x8a, 0x7f, 7, 0, 8, reset_tag_key)
dst = blocks.vector_sink_b()
self.tb.connect(src, scrambler, dst)
self.tb.run()
expected_data = additive_scramble_lfsr(0x8a, 0x7f, 7, 8, src_data)
self.assertEqual(expected_data * 3, dst.data())
if __name__ == '__main__':
gr_unittest.run(test_scrambler, "test_scrambler.xml")
|
