How to test source blocks in gnuradio

问题:

I'm trying to create a simple source block in gnuradio. I've used gr_modtool to create the bare bones module and block, but whenever I try to run the tests, it quickly eats up all my memory and my computer starts lagging. Even worse, the tests fail with "thread[thread-per-block[1]: ]: std::bad_alloc"

Here's my block (called csv):

import numpy
from gnuradio import gr

class csv(gr.sync_block):
    """
    docstring for block csv
    """
    def __init__(self, filename):
        gr.sync_block.__init__(self, name="csv",
                               in_sig=None,
                               out_sig=[numpy.float32])


    def work(self, input_items, output_items):
        out = output_items[0]
        out[:] = 0
        return len(output_items[0])

And here's the code I'm using to test:

from gnuradio import gr, gr_unittest
from gnuradio import blocks
from csv import csv

import time

class qa_csv (gr_unittest.TestCase):

    def setUp (self):
        self.tb = gr.top_block ()

    def tearDown (self):
        self.tb = None

    def test_001_t (self):
        expected = [0.0]

        # set up fg
        c = csv(None)
        sink = blocks.vector_sink_f()
        self.tb.connect(c, sink)

        self.tb.run()

        # check data
        results = sink.data()
        self.assertFloatTuplesAlmostEqual(expected, results)

if __name__ == '__main__':
    gr_unittest.run(qa_csv, "qa_csv.xml")

Can anyone help me figure out where I'm going wrong or point me in the right direction?

回答1:

A couple of things:

By default a flowgraph will run until it's told to stop. So that's why your memory is being chewed up. GNU Radio is throwing std::bad_alloc because you keep stuffing things in the vector_sink and eventually it runs out of RAM.

You need to stop the flowgraph. There are a couple ways to do that:

• Return WORK_DONE (-1) from the work function of the block. That's appropriate for when the block has a finite amount of data to serve and then signal it's done. Check out vector_source as an example.

• Use the head block, which will copy N samples and then return WORK_DONE for you. This is useful for unit tests.

Last note: once you get that working, your test will still fail (unless you request head to only copy 1 sample) because as you've written it, your source block will fill its entire output buffer with zeros each time it's called:

>>> import numpy as np
>>> out = np.empty(10, dtype=float)
>>> out[:] = 0
>>> out
array([ 0.,  0.,  0.,  0.,  0.,  0.,  0.,  0.,  0.,  0.])

so just make sure your expected array has the same number samples that you request from head:

>>> from gnuradio import gr, blocks
>>> n = 1000
>>> head = blocks.head(gr.sizeof_float, n)
>>> expected = np.zeros(n)