--- /dev/null
+#!/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()
+
+
+
+
+
+
projects / debian / gnuradio / blobdiff