filter: copied over pfb_channelizer to gr-filter with QA and GRC.

4 files changed, 180 insertions, 1 deletions

diff --git a/gr-filter/python/CMakeLists.txt b/gr-filter/python/CMakeLists.txt
index e52cac759..99a1aa3a9 100644
--- a/gr-filter/python/CMakeLists.txt
+++ b/gr-filter/python/CMakeLists.txt
@@ -23,6 +23,7 @@ include(GrPython)
 GR_PYTHON_INSTALL(
     FILES
     __init__.py
+    pfb.py
     DESTINATION ${GR_PYTHON_DIR}/gnuradio/filter
     COMPONENT "filter_python"
 )
diff --git a/gr-filter/python/__init__.py b/gr-filter/python/__init__.py
index 56dd2dc5a..90b5ce0a4 100644
--- a/gr-filter/python/__init__.py
+++ b/gr-filter/python/__init__.py
@@ -25,4 +25,4 @@ processing blocks for FILTER and related functions.
 '''
 
 from filter_swig import *
-
+import pfb
diff --git a/gr-filter/python/pfb.py b/gr-filter/python/pfb.py
new file mode 100644
index 000000000..9c7e18e31
--- /dev/null
+++ b/gr-filter/python/pfb.py
@@ -0,0 +1,74 @@
+#!/usr/bin/env python
+#
+# Copyright 2009,2010,2012 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio 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.
+#
+# GNU Radio 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 GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+from gnuradio import gr, optfir
+import filter_swig as filter
+
+class channelizer_ccf(gr.hier_block2):
+    '''
+    Make a Polyphase Filter channelizer (complex in, complex out, floating-point taps)
+
+    This simplifies the interface by allowing a single input stream to connect to this block.
+    It will then output a stream for each channel.
+    '''
+    def __init__(self, numchans, taps=None, oversample_rate=1, atten=100):
+	gr.hier_block2.__init__(self, "pfb_channelizer_ccf",
+				gr.io_signature(1, 1, gr.sizeof_gr_complex), # Input signature
+				gr.io_signature(numchans, numchans, gr.sizeof_gr_complex)) # Output signature
+
+        self._nchans = numchans
+        self._oversample_rate = oversample_rate
+
+        if taps is not None:
+            self._taps = taps
+        else:
+            # Create a filter that covers the full bandwidth of the input signal
+            bw = 0.4
+            tb = 0.2
+            ripple = 0.1
+            made = False
+            while not made:
+                try:
+                    self._taps = optfir.low_pass(1, self._nchans, bw, bw+tb, ripple, atten)
+                    made = True
+                except RuntimeError:
+                    ripple += 0.01
+                    made = False
+                    print("Warning: set ripple to %.4f dB. If this is a problem, adjust the attenuation or create your own filter taps." % (ripple))
+
+                    # Build in an exit strategy; if we've come this far, it ain't working.
+                    if(ripple >= 1.0):
+                        raise RuntimeError("optfir could not generate an appropriate filter.")
+
+        self.s2ss = gr.stream_to_streams(gr.sizeof_gr_complex, self._nchans)
+        self.pfb = filter.pfb_channelizer_ccf(self._nchans, self._taps,
+                                              self._oversample_rate)
+        self.connect(self, self.s2ss)
+
+        for i in xrange(self._nchans):
+            self.connect((self.s2ss,i), (self.pfb,i))
+            self.connect((self.pfb,i), (self,i))
+
+    def set_channel_map(self, newmap):
+        self.pfb.set_channel_map(newmap)
+
+
diff --git a/gr-filter/python/qa_pfb_channelizer.py b/gr-filter/python/qa_pfb_channelizer.py
new file mode 100644
index 000000000..b52c80e8b
--- /dev/null
+++ b/gr-filter/python/qa_pfb_channelizer.py
@@ -0,0 +1,104 @@
+#!/usr/bin/env python
+#
+# Copyright 2012 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio 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.
+#
+# GNU Radio 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 GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+from gnuradio import gr, gr_unittest
+import filter_swig as filter
+import math
+
+class test_pfb_channelizer(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gr.top_block()
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_000(self):
+        N = 1000         # number of samples to use
+        M = 5            # Number of channels to channelize
+        fs = 1000        # baseband sampling rate
+        ifs = M*fs       # input samp rate to channelizer
+
+        taps = gr.firdes.low_pass_2(1, ifs, 500, 50,
+                                    attenuation_dB=80,
+                                    window=gr.firdes.WIN_BLACKMAN_hARRIS)
+
+        signals = list()
+        add = gr.add_cc()
+        freqs = [-200, -100, 0, 100, 200]
+        for i in xrange(len(freqs)):
+            f = freqs[i] + (M/2-M+i+1)*fs
+            signals.append(gr.sig_source_c(ifs, gr.GR_SIN_WAVE, f, 1))
+            self.tb.connect(signals[i], (add,i))
+
+        head = gr.head(gr.sizeof_gr_complex, N)
+        s2ss = gr.stream_to_streams(gr.sizeof_gr_complex, M)
+        pfb = filter.pfb_channelizer_ccf(M, taps, 1)
+
+        self.tb.connect(add, head, s2ss)
+
+        snks = list()
+        for i in xrange(M):
+            snks.append(gr.vector_sink_c())
+            self.tb.connect((s2ss,i), (pfb,i))
+            self.tb.connect((pfb, i), snks[i])
+
+        self.tb.run() 
+
+        Ntest = 50
+        L = len(snks[0].data())
+        t = map(lambda x: float(x)/fs, xrange(L))
+
+        # Adjusted phase rotations for data
+        p0 = 0
+        p1 = 1.6335486
+        p2 = -3.01609
+        p3 = 3.01609
+        p4 = -1.6335486
+
+        # Create known data as complex sinusoids at the different baseband freqs
+        # the different channel numbering is due to channelizer output order.
+        expected0_data = map(lambda x: math.cos(2.*math.pi*freqs[2]*x+p0) + \
+                                       1j*math.sin(2.*math.pi*freqs[2]*x+p0), t)
+        expected1_data = map(lambda x: math.cos(2.*math.pi*freqs[3]*x+p1) + \
+                                       1j*math.sin(2.*math.pi*freqs[3]*x+p1), t)
+        expected2_data = map(lambda x: math.cos(2.*math.pi*freqs[4]*x+p2) + \
+                                       1j*math.sin(2.*math.pi*freqs[4]*x+p2), t)
+        expected3_data = map(lambda x: math.cos(2.*math.pi*freqs[0]*x+p3) + \
+                                       1j*math.sin(2.*math.pi*freqs[0]*x+p3), t)
+        expected4_data = map(lambda x: math.cos(2.*math.pi*freqs[1]*x+p4) + \
+                                       1j*math.sin(2.*math.pi*freqs[1]*x+p4), t)
+
+        dst0_data = snks[0].data()
+        dst1_data = snks[1].data()
+        dst2_data = snks[2].data()
+        dst3_data = snks[3].data()
+        dst4_data = snks[4].data()
+
+        self.assertComplexTuplesAlmostEqual(expected0_data[-Ntest:], dst0_data[-Ntest:], 4)
+        self.assertComplexTuplesAlmostEqual(expected1_data[-Ntest:], dst1_data[-Ntest:], 4)
+        self.assertComplexTuplesAlmostEqual(expected2_data[-Ntest:], dst2_data[-Ntest:], 4)
+        self.assertComplexTuplesAlmostEqual(expected3_data[-Ntest:], dst3_data[-Ntest:], 4)
+        self.assertComplexTuplesAlmostEqual(expected4_data[-Ntest:], dst4_data[-Ntest:], 4)
+
+if __name__ == '__main__':
+    gr_unittest.run(test_pfb_channelizer, "test_pfb_channelizer.xml")