diff options
Diffstat (limited to 'gr-atsc/src/python/qa_atsc.py')
| -rwxr-xr-x | gr-atsc/src/python/qa_atsc.py | 222 |
1 files changed, 0 insertions, 222 deletions
diff --git a/gr-atsc/src/python/qa_atsc.py b/gr-atsc/src/python/qa_atsc.py deleted file mode 100755 index c6b8d2e95..000000000 --- a/gr-atsc/src/python/qa_atsc.py +++ /dev/null @@ -1,222 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2004,2006,2007 Free Software Foundation, Inc. -# -# This file is part of GNU Radio -# -# GNU Radio is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation; either version 3, or (at your option) -# any later version. -# -# GNU Radio is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with GNU Radio; see the file COPYING. If not, write to -# the Free Software Foundation, Inc., 51 Franklin Street, -# Boston, MA 02110-1301, USA. -# - -from gnuradio import gr, gr_unittest -import atsc # qa code needs to run without being installed -#from gnuradio import atsc -from atsc_utils import * -import sys - - -class memoize(object): - def __init__(self, thunk): - self.thunk = thunk - self.cached = False - self.value = None - - def __call__(self): - if self.cached: - return self.value - self.value = self.thunk() - self.cached = True - return self.value - - -""" -Make a fake transport stream that's big enough for our purposes. -We generate 8 full fields. This is relatively expensive. It -takes about 2 seconds to execute. -""" -make_transport_stream = \ - memoize(lambda : tuple(make_fake_transport_stream_packet(8 * atsc.ATSC_DSEGS_PER_FIELD))) - - -def pad_transport_stream(src): - """ - An MPEG transport stream packet is 188 bytes long. Internally we use a packet - that is 256 bytes long to help with buffer alignment. This function adds the - appropriate trailing padding to convert each packet from 188 to 256 bytes. - """ - return pad_stream(src, atsc.sizeof_atsc_mpeg_packet, atsc.sizeof_atsc_mpeg_packet_pad) - - -def depad_transport_stream(src): - """ - An MPEG transport stream packet is 188 bytes long. Internally we use a packet - that is 256 bytes long to help with buffer alignment. This function removes the - trailing padding to convert each packet from 256 back to 188 bytes. - """ - return depad_stream(src, atsc.sizeof_atsc_mpeg_packet, atsc.sizeof_atsc_mpeg_packet_pad) - - -class vector_source_ts(gr.hier_block2): - """ - MPEG Transport stream source for testing. - """ - def __init__(self, ts): - """ - Pad tranport stream packets to 256 bytes and reformat appropriately. - - @param ts: MPEG transport stream. - @type ts: sequence of ints in [0,255]; len(ts) % 188 == 0 - """ - - src = gr.vector_source_b(pad_transport_stream(ts)) - s2v = gr.stream_to_vector(gr.sizeof_char, atsc.sizeof_atsc_mpeg_packet) - - gr.hier_block2.__init__(self, "vector_source_ts", - gr.io_signature(0, 0, 0), - s2v.output_signature()) - self.connect(src, s2v, self) - - -class vector_sink_ts(gr.hier_block2): - """ - MPEG Transport stream sink for testing. - """ - def __init__(self): - """ - """ - - v2s = gr.vector_to_stream(gr.sizeof_char, atsc.sizeof_atsc_mpeg_packet) - self.sink = gr.vector_sink_b() - gr.hier_block2.__init__(self, "vector_sink_ts", - v2s.input_signature(), - gr.io_signature(0, 0, 0)) - self.connect(self, v2s, self.sink) - - def data(self): - """ - Extracts tranport stream from sink and returns it to python. - - Depads tranport stream packets from 256 back to 188 bytes. - @rtype: tuple of ints in [0,255]; len(result) % 188 == 0 - """ - return tuple(depad_transport_stream(self.sink.data())) - - - -class qa_atsc(gr_unittest.TestCase): - - def setUp(self): - self.tb = gr.top_block() - - def tearDown(self): - self.tb = None - - - # The tests are run in alphabetical order - - def test_loopback_000(self): - """ - Loopback randomizer to derandomizer - """ - src_data = make_transport_stream() - expected_result = src_data - - src = vector_source_ts(src_data) - rand = atsc.randomizer() - derand = atsc.derandomizer() - dst = vector_sink_ts() - self.tb.connect(src, rand, derand, dst) - self.tb.run () - result_data = dst.data () - self.assertEqual (expected_result, result_data) - - def test_loopback_001(self): - """ - Loopback randomizer/rs_encoder to rs_decoder/derandomizer - """ - src_data = make_transport_stream() - expected_result = src_data - - src = vector_source_ts(src_data) - rand = atsc.randomizer() - rs_enc = atsc.rs_encoder() - rs_dec = atsc.rs_decoder() - derand = atsc.derandomizer() - dst = vector_sink_ts() - self.tb.connect(src, rand, rs_enc, rs_dec, derand, dst) - self.tb.run () - result_data = dst.data () - self.assertEqual (expected_result, result_data) - - def test_loopback_002(self): - """ - Loopback randomizer/rs_encoder/interleaver to - deinterleaver/rs_decoder/derandomizer - """ - src_data = make_transport_stream() - interleaver_delay = 52 - expected_result = src_data[0:len(src_data)-(interleaver_delay*atsc.ATSC_MPEG_PKT_LENGTH)] - - src = vector_source_ts(src_data) - rand = atsc.randomizer() - rs_enc = atsc.rs_encoder() - inter = atsc.interleaver() - deinter = atsc.deinterleaver() - rs_dec = atsc.rs_decoder() - derand = atsc.derandomizer() - dst = vector_sink_ts() - self.tb.connect(src, rand, rs_enc, inter, deinter, rs_dec, derand, dst) - self.tb.run () - result_data = dst.data () - result_data = result_data[(interleaver_delay*atsc.ATSC_MPEG_PKT_LENGTH):len(result_data)] - self.assertEqual (expected_result, result_data) - - - def test_loopback_003(self): - """ - Loopback randomizer/rs_encoder/interleaver/trellis_encoder - via ds_to_softds to - viterbi_decoder/deinterleaver/rs_decoder/derandomizer - """ - src_data = make_transport_stream() - interleaver_delay = 52 - viterbi_delay = 12 - expected_result = src_data[0:len(src_data)-((interleaver_delay+viterbi_delay)*atsc.ATSC_MPEG_PKT_LENGTH)] - - src = vector_source_ts(src_data) - rand = atsc.randomizer() - rs_enc = atsc.rs_encoder() - inter = atsc.interleaver() - trellis = atsc.trellis_encoder() - softds = atsc.ds_to_softds() - viterbi = atsc.viterbi_decoder() - deinter = atsc.deinterleaver() - rs_dec = atsc.rs_decoder() - derand = atsc.derandomizer() - dst = vector_sink_ts() - self.tb.connect(src, rand, rs_enc, inter, trellis, softds, viterbi, deinter, rs_dec, derand, dst) - self.tb.run () - result_data = dst.data ()[((interleaver_delay+viterbi_delay)*atsc.ATSC_MPEG_PKT_LENGTH):len(dst.data())] - self.assertEqual (expected_result, result_data) - - -if __name__ == '__main__': - gr_unittest.main() - - - - - - |