Can you spot the error in the python basic block?

GNU Radio
1

Yesterday I asked a question about a failing flowgraph when connecting
the copy block to a message source and running the flowgraph multiple
times.

I’m still trying to understand what’s going on, so I decided to write a
version of “copy” in python to try and better understand things. Even
though I’ve successfully written basic blocks in Python before, I am
really struggling to make this work.

As it’s implemented, it could (should) be a sync_block (and works if it
is), but copy needs to be a basic block to handle disabled copying.

The test hangs as if “py_copy” is never passing on the WORK_DONE from
the vector_sink. Does anyone see the error?

import numpy as np
from gnuradio import gr, gr_unittest, blocks

class py_copy(gr.basic_block):
“”“A python version of the copy block for comparison testing.”""
def init(self, itemtype):
gr.basic_block.init(
self,
name=“py_copy”,
in_sig=[itemtype],
out_sig=[itemtype]
)

def general_work(self, input_items, output_items):
    n = min(len(input_items[0]), len(output_items[0]))
    output_items[0][:n] = input_items[0][:n]
    self.consume_each(n)
    return n

class qa_copy(gr_unittest.TestCase):
def setUp(self):
self.tb = gr.top_block()

def tearDown(self):
    self.tb = None

def test_copy(self):
    src_data = np.arange(1000)
    src = blocks.vector_source_f(src_data)
    #copy_block = blocks.copy(gr.sizeof_float)
    copy_block = py_copy(np.float32)
    null_sink = blocks.null_sink(gr.sizeof_float)

    self.tb.connect(src, copy_block, null_sink)

    self.tb.run()
    self.assertEqual(1000, copy_block.nitems_written(0))

if name == ‘main’:
gr_unittest.run(qa_copy)

2

It seems like “consume_each” is not consuming the input buffer
correctly. I stepped through the general_work function with a debugger
and the input_items are being properly copied to output_items,
consume_each and return are both being passed the correct integer for
the number of input samples it got (1000), but the input buffer doesn’t
get consumed. Almost like vector_source_f was on “repeat”, but it’s not.
Even putting a head block after the vector source does nothing.

-Doug

From: discuss-gnuradio-bounces+danderson=removed_email_address@domain.invalid
[discuss-gnuradio-bounces+danderson=removed_email_address@domain.invalid] on behalf
of Anderson, Douglas J. [[email protected]]
Sent: Friday, March 13, 2015 12:26 PM
To: GNURadio D.ion List
Subject: [Discuss-gnuradio] Can you spot the error in the python basic
block?

Yesterday I asked a question about a failing flowgraph when connecting
the copy block to a message source and running the flowgraph multiple
times.

I’m still trying to understand what’s going on, so I decided to write a
version of “copy” in python to try and better understand things. Even
though I’ve successfully written basic blocks in Python before, I am
really struggling to make this work.

As it’s implemented, it could (should) be a sync_block (and works if it
is), but copy needs to be a basic block to handle disabled copying.

The test hangs as if “py_copy” is never passing on the WORK_DONE from
the vector_sink. Does anyone see the error?

import numpy as np
from gnuradio import gr, gr_unittest, blocks

class py_copy(gr.basic_block):
“”“A python version of the copy block for comparison testing.”""
def init(self, itemtype):
gr.basic_block.init(
self,
name=“py_copy”,
in_sig=[itemtype],
out_sig=[itemtype]
)

def general_work(self, input_items, output_items):
    n = min(len(input_items[0]), len(output_items[0]))
    output_items[0][:n] = input_items[0][:n]
    self.consume_each(n)
    return n

class qa_copy(gr_unittest.TestCase):
def setUp(self):
self.tb = gr.top_block()

def tearDown(self):
    self.tb = None

def test_copy(self):
    src_data = np.arange(1000)
    src = blocks.vector_source_f(src_data)
    #copy_block = blocks.copy(gr.sizeof_float)
    copy_block = py_copy(np.float32)
    null_sink = blocks.null_sink(gr.sizeof_float)

    self.tb.connect(src, copy_block, null_sink)

    self.tb.run()
    self.assertEqual(1000, copy_block.nitems_written(0))

if name == ‘main’:
gr_unittest.run(qa_copy)