[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
Re: [Discuss-gnuradio] Python block composed of other blocks
From: |
maiconkist |
Subject: |
Re: [Discuss-gnuradio] Python block composed of other blocks |
Date: |
Wed, 30 Jan 2013 12:31:36 -0800 (PST) |
I fixed the problem I had by changing in_sing for "input_signature".
But now my testbench never finish executing and print a value that is the 2x
the correct value.
My energy_detector code is as follows:
#####
import numpy as np
from gnuradio import gr, blks2
from gnuradio import fft
from gnuradio import uhd
class energy_detector(gr.hier_block2):
class energy_calculator(gr.sync_block):
""" Nested class
"""
def __init__(self, fft_size):
gr.sync_block.__init__(
self,
name = "energy_detector",
in_sig = [np.dtype((np.float32,
fft_size))],
out_sig = [np.float32]
)
def work(self, input_items, output_items):
"""
Energy calculator main function
"""
in0 = input_items[0][:]
out = output_items[0][:]
out[:] = np.sum( np.square(in0) )
print "energy = %f" % out[0]
self.consume_each(len(in0))
return len(out)
def __init__(self, fft_size):
"""
Constructor
"""
""" Initialize our energy detector block
in_sig: vector of fft_size elements
out_sig: one float for each input vector
"""
gr.hier_block2.__init__(
self,
name = "energy_detector",
input_signature = gr.io_signature(1, 1,
fft_size * gr.sizeof_float),
output_signature = gr.io_signature(1, 1,
gr.sizeof_float)
#in_sig = [np.dtype((np.float32, fft_size))],
#out_sig = [np.float32]
)
#
# Blocks
#
self.fft = fft.fft_vfc(fft_size, True, [], False)
self.v2s = gr.vector_to_stream(gr.sizeof_gr_complex, fft_size)
self.c2f = gr.complex_to_float()
self.s2v = gr.stream_to_vector(gr.sizeof_float, fft_size)
self.ec = self.energy_calculator(fft_size)
# Flow graph
self.connect(self, self.fft, self.v2s, self.c2f, self.s2v,
self.ec, self)
#####
And my testbench do this:
####
src_data = (1, 1)
expected_result = np.sum(np.square(src_data))
# blocks
src = gr.vector_source_f (src_data)
s2v = gr.stream_to_vector(gr.sizeof_float, len(src_data))
ed = energy_detector(len(src_data))
v2s = gr.vector_to_stream(gr.sizeof_float, 1)
dst = gr.vector_sink_f ()
# flowgraph
self.tb.connect (src, s2v, ed, v2s, dst)
self.tb.run ()
result_data = dst.data ()
self.assertEqual (expected_result, result_data)
####
Any suggestion ?
--
View this message in context:
http://gnuradio.4.n7.nabble.com/Python-block-composed-of-other-blocks-tp39279p39294.html
Sent from the GnuRadio mailing list archive at Nabble.com.