Re: [Discuss-gnuradio] Vector average with decimate block

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2018 <+YOU OR YOUR COMPANY+>.
#
# This is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3, or (at your option)
# any later version.
#
# This software is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this software; see the file COPYING. If not, write to
# the Free Software Foundation, Inc., 51 Franklin Street,
# Boston, MA 02110-1301, USA.
#

import numpy
from gnuradio import gr, gr_unittest
from gnuradio import blocks
from vave import vave

class qa_vave (gr_unittest.TestCase):

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

def tearDown (self):
self.tb = None

def test_001_t (self):
vsize = 1024
vdecimate = 4
vin = numpy.zeros(vsize*vdecimate)
for i in range(vsize):
vin[i] = float(i)
# create a set of vectors
src = blocks.vector_source_f( vin.tolist())
s2v = blocks.stream_to_vector(gr.sizeof_float, vsize)
# block we're testing
vblock = vave( vsize, vdecimate)

v2s = blocks.vector_to_stream( gr.sizeof_float, vsize)
snk = blocks.vector_sink_f(vsize)

self.tb.connect (src, s2v)
self.tb.connect (s2v, vblock)
self.tb.connect (vblock, snk)
# self.tb.connect (v2s, snk)
expected = vin[0:vsize]/4.
print 'Expected: ', expected[0:7]
outdata = None
waittime = 0.01

self.tb.run ()
outdata = snk.data()
print 'Output: ', outdata[0:7]
# check data
self.assertFloatTuplesAlmostEqual (expected, outdata, 6)

if __name__ == '__main__':
gr_unittest.run(qa_vave, "qa_vave.xml")
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2018 <+YOU OR YOUR COMPANY+>.
#
# This is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3, or (at your option)
# any later version.
#
# This software is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this software; see the file COPYING. If not, write to
# the Free Software Foundation, Inc., 51 Franklin Street,
# Boston, MA 02110-1301, USA.
#

import numpy
from gnuradio import gr

class vave(gr.decim_block):
"""
docstring for block vave
"""
def __init__(self, vsize, vdecimate):
gr.decim_block.__init__(self,
name="vave",
in_sig=[(numpy.float32, int(vsize))],
out_sig=[(numpy.float32, int(vsize))],
decim=int(vdecimate))
self.vsize = int(vsize)
self.vdecimate = int(vdecimate)
self.sum = numpy.zeros(self.vsize)
self.count = 0
self.oneovern = 1./float(self.vdecimate)

def forecast( self, noutput_items, ninput_items):
return self.vdecimate

def work(self, input_items, output_items):
"""
Work averages all input vectors and outputs one vector for all inputs
"""
out = output_items[0]
in0 = input_items[0]

# get the number of input vectors
n = len( input_items)
print 'Number of work inputs: ',n
nout = len( output_items)
print 'Number of work outputs: ',n

nv = len(in0)
for j in range(nv):
# get the lenght of one input
vin = in0[j]
print 'Length of input ', len(vin),' for vectors: ',j

# indicate consumption of a vector from input
self.consume(0, len(vin))

# now save this vector until all are received
self.sum = self.sum + vin
self.count = self.count + 1

if self.count >= self.vdecimate:
# normalize output average
out[:] = self.oneovern * self.sum[:]
# now reset the count and restart the sum
self.count = 0
self.sum = numpy.zeros( self.vdecimate)
return len(output_items[0])
# end for all input vectors
# if here, then not enough vectors input to produce an output
return 0
# end vave()

Hi Marcus,

Thanks for your suggestion. I was wandering around lost because I thought
input_items[0] was a list of vectors. Your hint made me realize it
is a list of inputs.

The list of vectors are accessed with

in0 = input_items[0]

To get an individual vector we need two steps

nv = len(in0)

to get the number of vectors and

for i in range(nv):
v = in0[i]

…

This is really, really not obvious in the guides (to me anyway).

The qa_vav test now passes

Now to make the block callable in grc …

Thanks again,

Glen