git.gag.com Git - debian/gnuradio/blob - gr-atsc/src/python/qa_atsc.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2004,2006,2007 Free Software Foundation, Inc.
   4 #
   5 # This file is part of GNU Radio
   6 #
   7 # GNU Radio is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3, or (at your option)
  10 # any later version.
  11 #
  12 # GNU Radio is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with GNU Radio; see the file COPYING.  If not, write to
  19 # the Free Software Foundation, Inc., 51 Franklin Street,
  20 # Boston, MA 02110-1301, USA.
  21 #
  22
  23 from gnuradio import gr, gr_unittest
  24 import atsc                    # qa code needs to run without being installed
  25 #from gnuradio import atsc
  26 from atsc_utils import *
  27 import sys
  28
  29
  30 class memoize(object):
  31     def __init__(self, thunk):
  32         self.thunk = thunk
  33         self.cached = False
  34         self.value = None
  35
  36     def __call__(self):
  37         if self.cached:
  38             return self.value
  39         self.value = self.thunk()
  40         self.cached = True
  41         return self.value
  42
  43
  44 """
  45 Make a fake transport stream that's big enough for our purposes.
  46 We generate 8 full fields.  This is relatively expensive.  It
  47 takes about 2 seconds to execute.
  48 """
  49 make_transport_stream = \
  50     memoize(lambda : tuple(make_fake_transport_stream_packet(8 * atsc.ATSC_DSEGS_PER_FIELD)))
  51
  52
  53 def pad_transport_stream(src):
  54     """
  55     An MPEG transport stream packet is 188 bytes long.  Internally we use a packet
  56     that is 256 bytes long to help with buffer alignment.  This function adds the
  57     appropriate trailing padding to convert each packet from 188 to 256 bytes.
  58     """
  59     return pad_stream(src, atsc.sizeof_atsc_mpeg_packet, atsc.sizeof_atsc_mpeg_packet_pad)
  60
  61
  62 def depad_transport_stream(src):
  63     """
  64     An MPEG transport stream packet is 188 bytes long.  Internally we use a packet
  65     that is 256 bytes long to help with buffer alignment.  This function removes the
  66     trailing padding to convert each packet from 256 back to 188 bytes.
  67     """
  68     return depad_stream(src, atsc.sizeof_atsc_mpeg_packet, atsc.sizeof_atsc_mpeg_packet_pad)
  69
  70
  71 class vector_source_ts(gr.hier_block2):
  72     """
  73     MPEG Transport stream source for testing.
  74     """
  75     def __init__(self, ts):
  76         """
  77         Pad tranport stream packets to 256 bytes and reformat appropriately.
  78
  79         @param ts: MPEG transport stream.
  80         @type  ts: sequence of ints in [0,255]; len(ts) % 188 == 0
  81         """
  82
  83         src = gr.vector_source_b(pad_transport_stream(ts))
  84         s2v = gr.stream_to_vector(gr.sizeof_char, atsc.sizeof_atsc_mpeg_packet)
  85
  86         gr.hier_block2.__init__(self, "vector_source_ts",
  87                                 gr.io_signature(0, 0, 0),
  88                                 s2v.output_signature())
  89         self.connect(src, s2v, self)
  90
  91
  92 class vector_sink_ts(gr.hier_block2):
  93     """
  94     MPEG Transport stream sink for testing.
  95     """
  96     def __init__(self):
  97         """
  98         """
  99
 100         v2s = gr.vector_to_stream(gr.sizeof_char, atsc.sizeof_atsc_mpeg_packet)
 101         self.sink = gr.vector_sink_b()
 102         gr.hier_block2.__init__(self, "vector_sink_ts",
 103                                 v2s.input_signature(),
 104                                 gr.io_signature(0, 0, 0))
 105         self.connect(self, v2s, self.sink)
 106
 107     def data(self):
 108         """
 109         Extracts tranport stream from sink and returns it to python.
 110
 111         Depads tranport stream packets from 256 back to 188 bytes.
 112         @rtype: tuple of ints in [0,255]; len(result) % 188 == 0
 113         """
 114         return tuple(depad_transport_stream(self.sink.data()))
 115
 116
 117
 118 class qa_atsc(gr_unittest.TestCase):
 119
 120     def setUp(self):
 121         self.tb = gr.top_block()
 122
 123     def tearDown(self):
 124         self.tb = None
 125
 126
 127     # The tests are run in alphabetical order
 128
 129     def test_loopback_000(self):
 130         """
 131         Loopback randomizer to derandomizer
 132         """
 133         src_data = make_transport_stream()
 134         expected_result = src_data
 135
 136         src = vector_source_ts(src_data)
 137         rand = atsc.randomizer()
 138         derand = atsc.derandomizer()
 139         dst = vector_sink_ts()
 140         self.tb.connect(src, rand, derand, dst)
 141         self.tb.run ()
 142         result_data = dst.data ()
 143         self.assertEqual (expected_result, result_data)
 144
 145     def test_loopback_001(self):
 146         """
 147         Loopback randomizer/rs_encoder to rs_decoder/derandomizer
 148         """
 149         src_data = make_transport_stream()
 150         expected_result = src_data
 151
 152         src = vector_source_ts(src_data)
 153         rand = atsc.randomizer()
 154         rs_enc = atsc.rs_encoder()
 155         rs_dec = atsc.rs_decoder()
 156         derand = atsc.derandomizer()
 157         dst = vector_sink_ts()
 158         self.tb.connect(src, rand, rs_enc, rs_dec, derand, dst)
 159         self.tb.run ()
 160         result_data = dst.data ()
 161         self.assertEqual (expected_result, result_data)
 162
 163     def test_loopback_002(self):
 164         """
 165         Loopback randomizer/rs_encoder/interleaver to
 166         deinterleaver/rs_decoder/derandomizer
 167         """
 168         src_data = make_transport_stream()
 169         interleaver_delay = 52
 170         expected_result = src_data[0:len(src_data)-(interleaver_delay*atsc.ATSC_MPEG_PKT_LENGTH)]
 171
 172         src = vector_source_ts(src_data)
 173         rand = atsc.randomizer()
 174         rs_enc = atsc.rs_encoder()
 175         inter = atsc.interleaver()
 176         deinter = atsc.deinterleaver()
 177         rs_dec = atsc.rs_decoder()
 178         derand = atsc.derandomizer()
 179         dst = vector_sink_ts()
 180         self.tb.connect(src, rand, rs_enc, inter, deinter, rs_dec, derand, dst)
 181         self.tb.run ()
 182         result_data = dst.data ()
 183         result_data = result_data[(interleaver_delay*atsc.ATSC_MPEG_PKT_LENGTH):len(result_data)]
 184         self.assertEqual (expected_result, result_data)
 185
 186
 187     def test_loopback_003(self):
 188         """
 189         Loopback randomizer/rs_encoder/interleaver/trellis_encoder
 190         via ds_to_softds to
 191         viterbi_decoder/deinterleaver/rs_decoder/derandomizer
 192         """
 193         src_data = make_transport_stream()
 194         interleaver_delay = 52
 195         viterbi_delay = 12
 196         expected_result = src_data[0:len(src_data)-((interleaver_delay+viterbi_delay)*atsc.ATSC_MPEG_PKT_LENGTH)]
 197
 198         src = vector_source_ts(src_data)
 199         rand = atsc.randomizer()
 200         rs_enc = atsc.rs_encoder()
 201         inter = atsc.interleaver()
 202         trellis = atsc.trellis_encoder()
 203         softds = atsc.ds_to_softds()
 204         viterbi = atsc.viterbi_decoder()
 205         deinter = atsc.deinterleaver()
 206         rs_dec = atsc.rs_decoder()
 207         derand = atsc.derandomizer()
 208         dst = vector_sink_ts()
 209         self.tb.connect(src, rand, rs_enc, inter, trellis, softds, viterbi, deinter, rs_dec, derand, dst)
 210         self.tb.run ()
 211         result_data = dst.data ()[((interleaver_delay+viterbi_delay)*atsc.ATSC_MPEG_PKT_LENGTH):len(dst.data())]
 212         self.assertEqual (expected_result, result_data)
 213
 214
 215 if __name__ == '__main__':
 216     gr_unittest.main()
 217
 218
 219
 220
 221
 222