# Can you spot the error in the python basic block?

**URL:** <https://www.ruby-forum.com/t/can-you-spot-the-error-in-the-python-basic-block/234565>\
**Category:** GNU Radio\
**Created:** [March 13, 2015, 7:39pm UTC](https://www.ruby-forum.com/t/can-you-spot-the-error-in-the-python-basic-block/234565 "2015-03-13T19:39:06Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![addis\_a](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/a/c4cdca/32.png) [@addis\_a](https://www.ruby-forum.com/u/addis_a)\
**Post date:** [March 13, 2015, 7:39pm UTC](https://www.ruby-forum.com/t/can-you-spot-the-error-in-the-python-basic-block/234565/1 "2015-03-13T19:39:06Z")

</div>

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)

---

<div class="post-metadata">

**Author:** ![AndersonS\_Douglas\_J](https://www.ruby-forum.com/letter_avatar_proxy/v4/letter/a/53a042/32.png) [@AndersonS\_Douglas\_J](https://www.ruby-forum.com/u/AndersonS_Douglas_J)\
**Post date:** [March 13, 2015, 9:07pm UTC](https://www.ruby-forum.com/t/can-you-spot-the-error-in-the-python-basic-block/234565/2 "2015-03-13T21:07:43Z")

</div>

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. [removed\_email\_address@domain.invalid]  
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)