diff options
author | Johnathan Corgan | 2012-06-20 07:58:00 -0700 |
---|---|---|
committer | Johnathan Corgan | 2012-06-20 07:58:00 -0700 |
commit | 2036030273d1c4842670ca3b2bd96e72aa408c9e (patch) | |
tree | 46d627b9a6114d3f2ac5c52f2d430d5bec800818 /gr-filter/python | |
parent | 439b427e80ebc767db6e4e732213d89c2a6febd4 (diff) | |
parent | 446494ea3bb021cb0b423c13bde2689c5046fe39 (diff) | |
download | gnuradio-2036030273d1c4842670ca3b2bd96e72aa408c9e.tar.gz gnuradio-2036030273d1c4842670ca3b2bd96e72aa408c9e.tar.bz2 gnuradio-2036030273d1c4842670ca3b2bd96e72aa408c9e.zip |
Merge branch 'master' into wip/gr-blocks-master
Diffstat (limited to 'gr-filter/python')
25 files changed, 4274 insertions, 0 deletions
diff --git a/gr-filter/python/CMakeLists.txt b/gr-filter/python/CMakeLists.txt new file mode 100644 index 000000000..07f03fac4 --- /dev/null +++ b/gr-filter/python/CMakeLists.txt @@ -0,0 +1,50 @@ +# 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. + +######################################################################## +include(GrPython) + +GR_PYTHON_INSTALL( + FILES + __init__.py + optfir.py + pfb.py + rational_resampler.py + DESTINATION ${GR_PYTHON_DIR}/gnuradio/filter + COMPONENT "filter_python" +) + +######################################################################## +# Handle the unit tests +######################################################################## +if(ENABLE_TESTING) +include(GrTest) +file(GLOB py_qa_test_files "qa_*.py") +foreach(py_qa_test_file ${py_qa_test_files}) + get_filename_component(py_qa_test_name ${py_qa_test_file} NAME_WE) + set(GR_TEST_PYTHON_DIRS + ${CMAKE_BINARY_DIR}/gnuradio-core/src/python + ${CMAKE_BINARY_DIR}/gnuradio-core/src/lib/swig + ${CMAKE_BINARY_DIR}/gr-filter/python + ${CMAKE_BINARY_DIR}/gr-filter/swig + ) + set(GR_TEST_TARGET_DEPS gruel gnuradio-core gnuradio-filter) + GR_ADD_TEST(${py_qa_test_name} ${PYTHON_EXECUTABLE} ${py_qa_test_file}) +endforeach(py_qa_test_file) +endif(ENABLE_TESTING) diff --git a/gr-filter/python/__init__.py b/gr-filter/python/__init__.py new file mode 100644 index 000000000..65a62d828 --- /dev/null +++ b/gr-filter/python/__init__.py @@ -0,0 +1,30 @@ +# +# 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. +# + +''' +This is the gr-filter package. This package provides GNU Radio +processing blocks for FILTER and related functions. +''' + +from filter_swig import * +from rational_resampler import * +import pfb +import optfir diff --git a/gr-filter/python/optfir.py b/gr-filter/python/optfir.py new file mode 100644 index 000000000..bccb8c68d --- /dev/null +++ b/gr-filter/python/optfir.py @@ -0,0 +1,339 @@ +# +# Copyright 2004,2005,2009 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. +# + +''' +Routines for designing optimal FIR filters. + +For a great intro to how all this stuff works, see section 6.6 of +"Digital Signal Processing: A Practical Approach", Emmanuael C. Ifeachor +and Barrie W. Jervis, Adison-Wesley, 1993. ISBN 0-201-54413-X. +''' + +import math, cmath +import filter_swig as filter + +# ---------------------------------------------------------------- + +## Builds a low pass filter. +# @param gain Filter gain in the passband (linear) +# @param Fs Sampling rate (sps) +# @param freq1 End of pass band (in Hz) +# @param freq2 Start of stop band (in Hz) +# @param passband_ripple_db Pass band ripple in dB (should be small, < 1) +# @param stopband_atten_db Stop band attenuation in dB (should be large, >= 60) +# @param nextra_taps Extra taps to use in the filter (default=2) +def low_pass (gain, Fs, freq1, freq2, passband_ripple_db, stopband_atten_db, + nextra_taps=2): + passband_dev = passband_ripple_to_dev (passband_ripple_db) + stopband_dev = stopband_atten_to_dev (stopband_atten_db) + desired_ampls = (gain, 0) + (n, fo, ao, w) = remezord ([freq1, freq2], desired_ampls, + [passband_dev, stopband_dev], Fs) + # The remezord typically under-estimates the filter order, so add 2 taps by default + taps = filter.pm_remez (n + nextra_taps, fo, ao, w, "bandpass") + return taps + +## Builds a band pass filter. +# @param gain Filter gain in the passband (linear) +# @param Fs Sampling rate (sps) +# @param freq_sb1 End of stop band (in Hz) +# @param freq_pb1 Start of pass band (in Hz) +# @param freq_pb2 End of pass band (in Hz) +# @param freq_sb2 Start of stop band (in Hz) +# @param passband_ripple_db Pass band ripple in dB (should be small, < 1) +# @param stopband_atten_db Stop band attenuation in dB (should be large, >= 60) +# @param nextra_taps Extra taps to use in the filter (default=2) +def band_pass (gain, Fs, freq_sb1, freq_pb1, freq_pb2, freq_sb2, + passband_ripple_db, stopband_atten_db, + nextra_taps=2): + passband_dev = passband_ripple_to_dev (passband_ripple_db) + stopband_dev = stopband_atten_to_dev (stopband_atten_db) + desired_ampls = (0, gain, 0) + desired_freqs = [freq_sb1, freq_pb1, freq_pb2, freq_sb2] + desired_ripple = [stopband_dev, passband_dev, stopband_dev] + (n, fo, ao, w) = remezord (desired_freqs, desired_ampls, + desired_ripple, Fs) + # The remezord typically under-estimates the filter order, so add 2 taps by default + taps = filter.pm_remez (n + nextra_taps, fo, ao, w, "bandpass") + return taps + + +## Builds a band pass filter with complex taps by making an LPF and +# spinning it up to the right center frequency +# @param gain Filter gain in the passband (linear) +# @param Fs Sampling rate (sps) +# @param freq_sb1 End of stop band (in Hz) +# @param freq_pb1 Start of pass band (in Hz) +# @param freq_pb2 End of pass band (in Hz) +# @param freq_sb2 Start of stop band (in Hz) +# @param passband_ripple_db Pass band ripple in dB (should be small, < 1) +# @param stopband_atten_db Stop band attenuation in dB (should be large, >= 60) +# @param nextra_taps Extra taps to use in the filter (default=2) +def complex_band_pass (gain, Fs, freq_sb1, freq_pb1, freq_pb2, freq_sb2, + passband_ripple_db, stopband_atten_db, + nextra_taps=2): + center_freq = (freq_pb2 + freq_pb1) / 2.0 + lp_pb = (freq_pb2 - center_freq)/1.0 + lp_sb = freq_sb2 - center_freq + lptaps = low_pass(gain, Fs, lp_pb, lp_sb, passband_ripple_db, + stopband_atten_db, nextra_taps) + spinner = [cmath.exp(2j*cmath.pi*center_freq/Fs*i) for i in xrange(len(lptaps))] + taps = [s*t for s,t in zip(spinner, lptaps)] + return taps + + +## Builds a band reject filter +# spinning it up to the right center frequency +# @param gain Filter gain in the passband (linear) +# @param Fs Sampling rate (sps) +# @param freq_pb1 End of pass band (in Hz) +# @param freq_sb1 Start of stop band (in Hz) +# @param freq_sb2 End of stop band (in Hz) +# @param freq_pb2 Start of pass band (in Hz) +# @param passband_ripple_db Pass band ripple in dB (should be small, < 1) +# @param stopband_atten_db Stop band attenuation in dB (should be large, >= 60) +# @param nextra_taps Extra taps to use in the filter (default=2) +def band_reject (gain, Fs, freq_pb1, freq_sb1, freq_sb2, freq_pb2, + passband_ripple_db, stopband_atten_db, + nextra_taps=2): + passband_dev = passband_ripple_to_dev (passband_ripple_db) + stopband_dev = stopband_atten_to_dev (stopband_atten_db) + desired_ampls = (gain, 0, gain) + desired_freqs = [freq_pb1, freq_sb1, freq_sb2, freq_pb2] + desired_ripple = [passband_dev, stopband_dev, passband_dev] + (n, fo, ao, w) = remezord (desired_freqs, desired_ampls, + desired_ripple, Fs) + # Make sure we use an odd number of taps + if((n+nextra_taps)%2 == 1): + n += 1 + # The remezord typically under-estimates the filter order, so add 2 taps by default + taps = filter.pm_remez (n + nextra_taps, fo, ao, w, "bandpass") + return taps + + +## Builds a high pass filter. +# @param gain Filter gain in the passband (linear) +# @param Fs Sampling rate (sps) +# @param freq1 End of stop band (in Hz) +# @param freq2 Start of pass band (in Hz) +# @param passband_ripple_db Pass band ripple in dB (should be small, < 1) +# @param stopband_atten_db Stop band attenuation in dB (should be large, >= 60) +# @param nextra_taps Extra taps to use in the filter (default=2) +def high_pass (gain, Fs, freq1, freq2, passband_ripple_db, stopband_atten_db, + nextra_taps=2): + passband_dev = passband_ripple_to_dev (passband_ripple_db) + stopband_dev = stopband_atten_to_dev (stopband_atten_db) + desired_ampls = (0, 1) + (n, fo, ao, w) = remezord ([freq1, freq2], desired_ampls, + [stopband_dev, passband_dev], Fs) + # For a HPF, we need to use an odd number of taps + # In filter.remez, ntaps = n+1, so n must be even + if((n+nextra_taps)%2 == 1): + n += 1 + + # The remezord typically under-estimates the filter order, so add 2 taps by default + taps = filter.pm_remez (n + nextra_taps, fo, ao, w, "bandpass") + return taps + +# ---------------------------------------------------------------- + +def stopband_atten_to_dev (atten_db): + """Convert a stopband attenuation in dB to an absolute value""" + return 10**(-atten_db/20) + +def passband_ripple_to_dev (ripple_db): + """Convert passband ripple spec expressed in dB to an absolute value""" + return (10**(ripple_db/20)-1)/(10**(ripple_db/20)+1) + +# ---------------------------------------------------------------- + +def remezord (fcuts, mags, devs, fsamp = 2): + ''' + FIR order estimator (lowpass, highpass, bandpass, mulitiband). + + (n, fo, ao, w) = remezord (f, a, dev) + (n, fo, ao, w) = remezord (f, a, dev, fs) + + (n, fo, ao, w) = remezord (f, a, dev) finds the approximate order, + normalized frequency band edges, frequency band amplitudes, and + weights that meet input specifications f, a, and dev, to use with + the remez command. + + * f is a sequence of frequency band edges (between 0 and Fs/2, where + Fs is the sampling frequency), and a is a sequence specifying the + desired amplitude on the bands defined by f. The length of f is + twice the length of a, minus 2. The desired function is + piecewise constant. + + * dev is a sequence the same size as a that specifies the maximum + allowable deviation or ripples between the frequency response + and the desired amplitude of the output filter, for each band. + + Use remez with the resulting order n, frequency sequence fo, + amplitude response sequence ao, and weights w to design the filter b + which approximately meets the specifications given by remezord + input parameters f, a, and dev: + + b = remez (n, fo, ao, w) + + (n, fo, ao, w) = remezord (f, a, dev, Fs) specifies a sampling frequency Fs. + + Fs defaults to 2 Hz, implying a Nyquist frequency of 1 Hz. You can + therefore specify band edges scaled to a particular applications + sampling frequency. + + In some cases remezord underestimates the order n. If the filter + does not meet the specifications, try a higher order such as n+1 + or n+2. + ''' + # get local copies + fcuts = fcuts[:] + mags = mags[:] + devs = devs[:] + + for i in range (len (fcuts)): + fcuts[i] = float (fcuts[i]) / fsamp + + nf = len (fcuts) + nm = len (mags) + nd = len (devs) + nbands = nm + + if nm != nd: + raise ValueError, "Length of mags and devs must be equal" + + if nf != 2 * (nbands - 1): + raise ValueError, "Length of f must be 2 * len (mags) - 2" + + for i in range (len (mags)): + if mags[i] != 0: # if not stopband, get relative deviation + devs[i] = devs[i] / mags[i] + + # separate the passband and stopband edges + f1 = fcuts[0::2] + f2 = fcuts[1::2] + + n = 0 + min_delta = 2 + for i in range (len (f1)): + if f2[i] - f1[i] < min_delta: + n = i + min_delta = f2[i] - f1[i] + + if nbands == 2: + # lowpass or highpass case (use formula) + l = lporder (f1[n], f2[n], devs[0], devs[1]) + else: + # bandpass or multipass case + # try different lowpasses and take the worst one that + # goes through the BP specs + l = 0 + for i in range (1, nbands-1): + l1 = lporder (f1[i-1], f2[i-1], devs[i], devs[i-1]) + l2 = lporder (f1[i], f2[i], devs[i], devs[i+1]) + l = max (l, l1, l2) + + n = int (math.ceil (l)) - 1 # need order, not length for remez + + # cook up remez compatible result + ff = [0] + fcuts + [1] + for i in range (1, len (ff) - 1): + ff[i] *= 2 + + aa = [] + for a in mags: + aa = aa + [a, a] + + max_dev = max (devs) + wts = [1] * len(devs) + for i in range (len (wts)): + wts[i] = max_dev / devs[i] + + return (n, ff, aa, wts) + +# ---------------------------------------------------------------- + +def lporder (freq1, freq2, delta_p, delta_s): + ''' + FIR lowpass filter length estimator. freq1 and freq2 are + normalized to the sampling frequency. delta_p is the passband + deviation (ripple), delta_s is the stopband deviation (ripple). + + Note, this works for high pass filters too (freq1 > freq2), but + doesnt work well if the transition is near f == 0 or f == fs/2 + + From Herrmann et al (1973), Practical design rules for optimum + finite impulse response filters. Bell System Technical J., 52, 769-99 + ''' + df = abs (freq2 - freq1) + ddp = math.log10 (delta_p) + dds = math.log10 (delta_s) + + a1 = 5.309e-3 + a2 = 7.114e-2 + a3 = -4.761e-1 + a4 = -2.66e-3 + a5 = -5.941e-1 + a6 = -4.278e-1 + + b1 = 11.01217 + b2 = 0.5124401 + + t1 = a1 * ddp * ddp + t2 = a2 * ddp + t3 = a4 * ddp * ddp + t4 = a5 * ddp + + dinf=((t1 + t2 + a3) * dds) + (t3 + t4 + a6) + ff = b1 + b2 * (ddp - dds) + n = dinf / df - ff * df + 1 + return n + + +def bporder (freq1, freq2, delta_p, delta_s): + ''' + FIR bandpass filter length estimator. freq1 and freq2 are + normalized to the sampling frequency. delta_p is the passband + deviation (ripple), delta_s is the stopband deviation (ripple). + + From Mintzer and Liu (1979) + ''' + df = abs (freq2 - freq1) + ddp = math.log10 (delta_p) + dds = math.log10 (delta_s) + + a1 = 0.01201 + a2 = 0.09664 + a3 = -0.51325 + a4 = 0.00203 + a5 = -0.57054 + a6 = -0.44314 + + t1 = a1 * ddp * ddp + t2 = a2 * ddp + t3 = a4 * ddp * ddp + t4 = a5 * ddp + + cinf = dds * (t1 + t2 + a3) + t3 + t4 + a6 + ginf = -14.6 * math.log10 (delta_p / delta_s) - 16.9 + n = cinf / df + ginf * df + 1 + return n + diff --git a/gr-filter/python/pfb.py b/gr-filter/python/pfb.py new file mode 100644 index 000000000..ddf289982 --- /dev/null +++ b/gr-filter/python/pfb.py @@ -0,0 +1,271 @@ +#!/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 +import filter_swig as filter +import optfir + +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), + gr.io_signature(numchans, numchans, gr.sizeof_gr_complex)) + + 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) + + + +class interpolator_ccf(gr.hier_block2): + ''' + Make a Polyphase Filter interpolator (complex in, complex out, floating-point taps) + + The block takes a single complex stream in and outputs a single complex + stream out. As such, it requires no extra glue to handle the input/output + streams. This block is provided to be consistent with the interface to the + other PFB block. + ''' + def __init__(self, interp, taps=None, atten=100): + gr.hier_block2.__init__(self, "pfb_interpolator_ccf", + gr.io_signature(1, 1, gr.sizeof_gr_complex), + gr.io_signature(1, 1, gr.sizeof_gr_complex)) + + self._interp = interp + self._taps = taps + + 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.99 + made = False + while not made: + try: + self._taps = optfir.low_pass(self._interp, self._interp, 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.pfb = filter.pfb_interpolator_ccf(self._interp, self._taps) + + self.connect(self, self.pfb) + self.connect(self.pfb, self) + + +class decimator_ccf(gr.hier_block2): + ''' + Make a Polyphase Filter decimator (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 that is the decimated output stream. + ''' + def __init__(self, decim, taps=None, channel=0, atten=100): + gr.hier_block2.__init__(self, "pfb_decimator_ccf", + gr.io_signature(1, 1, gr.sizeof_gr_complex), + gr.io_signature(1, 1, gr.sizeof_gr_complex)) + + self._decim = decim + self._channel = channel + + 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._decim, 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._decim) + self.pfb = filter.pfb_decimator_ccf(self._decim, self._taps, self._channel) + + self.connect(self, self.s2ss) + + for i in xrange(self._decim): + self.connect((self.s2ss,i), (self.pfb,i)) + + self.connect(self.pfb, self) + + +class arb_resampler_ccf(gr.hier_block2): + ''' + Convenience wrapper for the polyphase filterbank arbitrary resampler. + + The block takes a single complex stream in and outputs a single complex + stream out. As such, it requires no extra glue to handle the input/output + streams. This block is provided to be consistent with the interface to the + other PFB block. + ''' + def __init__(self, rate, taps=None, flt_size=32, atten=100): + gr.hier_block2.__init__(self, "pfb_arb_resampler_ccf", + gr.io_signature(1, 1, gr.sizeof_gr_complex), # Input signature + gr.io_signature(1, 1, gr.sizeof_gr_complex)) # Output signature + + self._rate = rate + self._size = flt_size + + 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 + #self._taps = filter.firdes.low_pass_2(self._size, self._size, bw, tb, atten) + made = False + while not made: + try: + self._taps = optfir.low_pass(self._size, self._size, 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.pfb = filter.pfb_arb_resampler_ccf(self._rate, self._taps, self._size) + #print "PFB has %d taps\n" % (len(self._taps),) + + self.connect(self, self.pfb) + self.connect(self.pfb, self) + + # Note -- set_taps not implemented in base class yet + def set_taps(self, taps): + self.pfb.set_taps(taps) + + def set_rate(self, rate): + self.pfb.set_rate(rate) + + +class arb_resampler_fff(gr.hier_block2): + ''' + Convenience wrapper for the polyphase filterbank arbitrary resampler. + + The block takes a single float stream in and outputs a single float + stream out. As such, it requires no extra glue to handle the input/output + streams. This block is provided to be consistent with the interface to the + other PFB block. + ''' + def __init__(self, rate, taps=None, flt_size=32, atten=100): + gr.hier_block2.__init__(self, "pfb_arb_resampler_fff", + gr.io_signature(1, 1, gr.sizeof_float), # Input signature + gr.io_signature(1, 1, gr.sizeof_float)) # Output signature + + self._rate = rate + self._size = flt_size + + 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 + #self._taps = filter.firdes.low_pass_2(self._size, self._size, bw, tb, atten) + made = False + while not made: + try: + self._taps = optfir.low_pass(self._size, self._size, 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.pfb = filter.pfb_arb_resampler_fff(self._rate, self._taps, self._size) + #print "PFB has %d taps\n" % (len(self._taps),) + + self.connect(self, self.pfb) + self.connect(self.pfb, self) + + # Note -- set_taps not implemented in base class yet + def set_taps(self, taps): + self.pfb.set_taps(taps) + + def set_rate(self, rate): + self.pfb.set_rate(rate) diff --git a/gr-filter/python/qa_adaptive_fir_filter.py b/gr-filter/python/qa_adaptive_fir_filter.py new file mode 100755 index 000000000..cadce5204 --- /dev/null +++ b/gr-filter/python/qa_adaptive_fir_filter.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python +# +# Copyright 2008,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 this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +# + +from gnuradio import gr, gr_unittest +import filter_swig as filter + +class test_adaptive_filter(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block () + + def tearDown(self): + self.tb = None + + def test_adaptive_fir_filter_ccf_001(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((0.5+0.5j), (1.5+1.5j), (3+3j), (5+5j), (5.5+5.5j), + (6.5+6.5j), (8+8j), (10+10j), (10.5+10.5j), (11.5+11.5j), + (13+13j), (15+15j), (15.5+15.5j), (16.5+16.5j), (18+18j), + (20+20j), (20.5+20.5j), (21.5+21.5j), (23+23j), (25+25j), + (25.5+25.5j), (26.5+26.5j), (28+28j), (30+30j), (30.5+30.5j), + (31.5+31.5j), (33+33j), (35+35j), (35.5+35.5j), (36.5+36.5j), + (38+38j), (40+40j), (40.5+40.5j), (41.5+41.5j), (43+43j), + (45+45j), (45.5+45.5j), (46.5+46.5j), (48+48j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j)) + + src = gr.vector_source_c(src_data) + op = filter.adaptive_fir_ccf("test", 1, 20*[0.5, 0.5]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_adaptive_fir_filter_ccf_002(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((0.5+0.5j), (5.5+5.5j), (10.5+10.5j), (15.5+15.5j), + (20.5+20.5j), (25.5+25.5j), (30.5+30.5j), (35.5+35.5j), + (40.5+40.5j), (45.5+45.5j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j)) + + src = gr.vector_source_c(src_data) + op = filter.adaptive_fir_ccf("test", 4, 20*[0.5, 0.5]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_adaptive_fir_filter_ccc_001(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((-0.5+1.5j), (-1.5+4.5j), (-3+9j), (-5+15j), + (-5.5+16.5j), (-6.5+19.5j), (-8+24j), (-10+30j), + (-10.5+31.5j), (-11.5+34.5j), (-13+39j), (-15+45j), + (-15.5+46.5j), (-16.5+49.5j), (-18+54j), (-20+60j), + (-20.5+61.5j), (-21.5+64.5j), (-23+69j), (-25+75j), + (-25.5+76.5j), (-26.5+79.5j), (-28+84j), (-30+90j), + (-30.5+91.5j), (-31.5+94.5j), (-33+99j), (-35+105j), + (-35.5+106.5j), (-36.5+109.5j), (-38+114j), (-40+120j), + (-40.5+121.5j), (-41.5+124.5j), (-43+129j), (-45+135j), + (-45.5+136.5j), (-46.5+139.5j), (-48+144j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j)) + src = gr.vector_source_c(src_data) + op = filter.adaptive_fir_ccc("test", 1, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + + def test_adaptive_fir_filter_ccc_002(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((-0.5+1.5j), (-5.5+16.5j), (-10.5+31.5j), + (-15.5+46.5j), (-20.5+61.5j), (-25.5+76.5j), + (-30.5+91.5j), (-35.5+106.5j), (-40.5+121.5j), + (-45.5+136.5j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j)) + src = gr.vector_source_c(src_data) + op = filter.adaptive_fir_ccc("test", 4, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + +if __name__ == '__main__': + gr_unittest.run(test_adaptive_filter, "test_adaptive_filter.xml") + diff --git a/gr-filter/python/qa_channel_model.py b/gr-filter/python/qa_channel_model.py new file mode 100755 index 000000000..7f1c61b4e --- /dev/null +++ b/gr-filter/python/qa_channel_model.py @@ -0,0 +1,60 @@ +#!/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_channel_model(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 + fs = 1000 # baseband sampling rate + freq = 100 + + signal = gr.sig_source_c(fs, gr.GR_SIN_WAVE, freq, 1) + head = gr.head(gr.sizeof_gr_complex, N) + op = filter.channel_model(0.0, 0.0, 1.0, [1,], 0) + snk = gr.vector_sink_c() + snk1 = gr.vector_sink_c() + + op.set_noise_voltage(0.0) + op.set_frequency_offset(0.0) + op.set_taps([1,]) + op.set_timing_offset(1.0) + + self.tb.connect(signal, head, op, snk) + self.tb.connect(op, snk1) + self.tb.run() + + dst_data = snk.data() + exp_data = snk1.data() + self.assertComplexTuplesAlmostEqual(exp_data, dst_data, 5) + +if __name__ == '__main__': + gr_unittest.run(test_channel_model, "test_channel_model.xml") diff --git a/gr-filter/python/qa_dc_blocker.py b/gr-filter/python/qa_dc_blocker.py new file mode 100755 index 000000000..b9df06b45 --- /dev/null +++ b/gr-filter/python/qa_dc_blocker.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python +# +# Copyright 2011,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 + +class test_dc_blocker(gr_unittest.TestCase): + + def setUp (self): + self.tb = gr.top_block () + + def tearDown (self): + self.tb = None + + def test_001(self): + ''' Test impulse response - long form, cc ''' + src_data = [1,] + 100*[0,] + expected_result = ((-0.02072429656982422+0j), (-0.02081298828125+0j), + (0.979156494140625+0j), (-0.02081298828125+0j), + (-0.02072429656982422+0j)) + + src = gr.vector_source_c(src_data) + op = filter.dc_blocker_cc(32, True) + dst = gr.vector_sink_c() + + self.tb.connect (src, op, dst) + self.tb.run() + + # only test samples around 2D-2 + result_data = dst.data()[60:65] + self.assertComplexTuplesAlmostEqual (expected_result, result_data) + + def test_002(self): + ''' Test impulse response - short form, cc ''' + src_data = [1,] + 100*[0,] + expected_result = ((-0.029296875+0j), (-0.0302734375+0j), + (0.96875+0j), (-0.0302734375+0j), + (-0.029296875+0j)) + + src = gr.vector_source_c(src_data) + op = filter.dc_blocker_cc(32, False) + dst = gr.vector_sink_c() + + self.tb.connect (src, op, dst) + self.tb.run() + + # only test samples around D-1 + result_data = dst.data()[29:34] + self.assertComplexTuplesAlmostEqual (expected_result, result_data) + + + def test_003(self): + ''' Test impulse response - long form, ff ''' + src_data = [1,] + 100*[0,] + expected_result = ((-0.02072429656982422), (-0.02081298828125), + (0.979156494140625), (-0.02081298828125), + (-0.02072429656982422)) + + src = gr.vector_source_f(src_data) + op = filter.dc_blocker_ff(32, True) + dst = gr.vector_sink_f() + + self.tb.connect (src, op, dst) + self.tb.run() + + # only test samples around 2D-2 + result_data = dst.data()[60:65] + self.assertFloatTuplesAlmostEqual (expected_result, result_data) + + def test_004(self): + ''' Test impulse response - short form, ff ''' + src_data = [1,] + 100*[0,] + expected_result = ((-0.029296875), (-0.0302734375), + (0.96875), (-0.0302734375), + (-0.029296875)) + + src = gr.vector_source_f(src_data) + op = filter.dc_blocker_ff(32, False) + dst = gr.vector_sink_f() + + self.tb.connect (src, op, dst) + self.tb.run() + + # only test samples around D-1 + result_data = dst.data()[29:34] + self.assertFloatTuplesAlmostEqual (expected_result, result_data) + +if __name__ == '__main__': + gr_unittest.run(test_dc_blocker, "test_dc_blocker.xml") + diff --git a/gr-filter/python/qa_fft_filter.py b/gr-filter/python/qa_fft_filter.py new file mode 100755 index 000000000..eaef3156d --- /dev/null +++ b/gr-filter/python/qa_fft_filter.py @@ -0,0 +1,380 @@ +#!/usr/bin/env python +# +# Copyright 2004,2005,2007,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, gr_unittest +import filter_swig as filter +import sys +import random + +def make_random_complex_tuple(L): + result = [] + for x in range(L): + result.append(complex(random.uniform(-1000,1000), + random.uniform(-1000,1000))) + return tuple(result) + +def make_random_float_tuple(L): + result = [] + for x in range(L): + result.append(float(int(random.uniform(-1000,1000)))) + return tuple(result) + + +def reference_filter_ccc(dec, taps, input): + """ + compute result using conventional fir filter + """ + tb = gr.top_block() + #src = gr.vector_source_c(((0,) * (len(taps) - 1)) + input) + src = gr.vector_source_c(input) + op = filter.fir_filter_ccc(dec, taps) + dst = gr.vector_sink_c() + tb.connect(src, op, dst) + tb.run() + return dst.data() + +def reference_filter_fff(dec, taps, input): + """ + compute result using conventional fir filter + """ + tb = gr.top_block() + #src = gr.vector_source_f(((0,) * (len(taps) - 1)) + input) + src = gr.vector_source_f(input) + op = filter.fir_filter_fff(dec, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + return dst.data() + + +def print_complex(x): + for i in x: + i = complex(i) + sys.stdout.write("(%6.3f,%6.3fj), " % (i.real, i.imag)) + sys.stdout.write('\n') + + +class test_fft_filter(gr_unittest.TestCase): + + def setUp(self): + pass + + def tearDown(self): + pass + + def assert_fft_ok2(self, expected_result, result_data): + expected_result = expected_result[:len(result_data)] + self.assertComplexTuplesAlmostEqual2 (expected_result, result_data, + abs_eps=1e-9, rel_eps=4e-4) + + def assert_fft_float_ok2(self, expected_result, result_data, abs_eps=1e-9, rel_eps=4e-4): + expected_result = expected_result[:len(result_data)] + self.assertFloatTuplesAlmostEqual2 (expected_result, result_data, + abs_eps, rel_eps) + + def test_ccc_001(self): + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (1,) + expected_result = tuple([complex(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(1, taps) + dst = gr.vector_sink_c() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + #print 'expected:', expected_result + #print 'results: ', result_data + self.assertComplexTuplesAlmostEqual (expected_result, result_data, 5) + + + def test_ccc_002(self): + # Test nthreads + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (2,) + nthreads = 2 + expected_result = tuple([2 * complex(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(1, taps, nthreads) + dst = gr.vector_sink_c() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + #print 'expected:', expected_result + #print 'results: ', result_data + self.assertComplexTuplesAlmostEqual (expected_result, result_data, 5) + + def test_ccc_003(self): + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (2,) + expected_result = tuple([2 * complex(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(1, taps) + dst = gr.vector_sink_c() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + #print 'expected:', expected_result + #print 'results: ', result_data + self.assertComplexTuplesAlmostEqual (expected_result, result_data, 5) + + + def test_ccc_004(self): + random.seed(0) + for i in xrange(25): + # sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + src_len = 4*1024 + src_data = make_random_complex_tuple(src_len) + ntaps = int(random.uniform(2, 1000)) + taps = make_random_complex_tuple(ntaps) + expected_result = reference_filter_ccc(1, taps, src_data) + + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(1, taps) + dst = gr.vector_sink_c() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + del tb + self.assert_fft_ok2(expected_result, result_data) + + def test_ccc_005(self): + random.seed(0) + for i in xrange(25): + # sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + dec = i + 1 + src_len = 4*1024 + src_data = make_random_complex_tuple(src_len) + ntaps = int(random.uniform(2, 100)) + taps = make_random_complex_tuple(ntaps) + expected_result = reference_filter_ccc(dec, taps, src_data) + + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(dec, taps) + dst = gr.vector_sink_c() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + del tb + result_data = dst.data() + + self.assert_fft_ok2(expected_result, result_data) + + def test_ccc_006(self): + # Test decimating with nthreads=2 + random.seed(0) + nthreads = 2 + for i in xrange(25): + # sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + dec = i + 1 + src_len = 4*1024 + src_data = make_random_complex_tuple(src_len) + ntaps = int(random.uniform(2, 100)) + taps = make_random_complex_tuple(ntaps) + expected_result = reference_filter_ccc(dec, taps, src_data) + + src = gr.vector_source_c(src_data) + op = filter.fft_filter_ccc(dec, taps, nthreads) + dst = gr.vector_sink_c() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + del tb + result_data = dst.data() + + self.assert_fft_ok2(expected_result, result_data) + + # ---------------------------------------------------------------- + # test _fff version + # ---------------------------------------------------------------- + + def test_fff_001(self): + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (1,) + expected_result = tuple([float(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + #print 'expected:', expected_result + #print 'results: ', result_data + self.assertFloatTuplesAlmostEqual (expected_result, result_data, 5) + + + def test_fff_002(self): + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (2,) + expected_result = tuple([2 * float(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + #print 'expected:', expected_result + #print 'results: ', result_data + self.assertFloatTuplesAlmostEqual (expected_result, result_data, 5) + + def test_fff_003(self): + # Test 02 with nthreads + tb = gr.top_block() + src_data = (0,1,2,3,4,5,6,7) + taps = (2,) + nthreads = 2 + expected_result = tuple([2 * float(x) for x in (0,1,2,3,4,5,6,7)]) + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(1, taps, nthreads) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual (expected_result, result_data, 5) + + def xtest_fff_004(self): + random.seed(0) + for i in xrange(25): + sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + src_len = 4096 + src_data = make_random_float_tuple(src_len) + ntaps = int(random.uniform(2, 1000)) + taps = make_random_float_tuple(ntaps) + expected_result = reference_filter_fff(1, taps, src_data) + + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(1, taps) + dst = gr.vector_sink_f() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + + #print "src_len =", src_len, " ntaps =", ntaps + try: + self.assert_fft_float_ok2(expected_result, result_data, abs_eps=1.0) + except: + expected = open('expected', 'w') + for x in expected_result: + expected.write(`x` + '\n') + actual = open('actual', 'w') + for x in result_data: + actual.write(`x` + '\n') + raise + + def xtest_fff_005(self): + random.seed(0) + for i in xrange(25): + sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + src_len = 4*1024 + src_data = make_random_float_tuple(src_len) + ntaps = int(random.uniform(2, 1000)) + taps = make_random_float_tuple(ntaps) + expected_result = reference_filter_fff(1, taps, src_data) + + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(1, taps) + dst = gr.vector_sink_f() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + + self.assert_fft_float_ok2(expected_result, result_data, abs_eps=2.0) + + def xtest_fff_006(self): + random.seed(0) + for i in xrange(25): + sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + dec = i + 1 + src_len = 4*1024 + src_data = make_random_float_tuple(src_len) + ntaps = int(random.uniform(2, 100)) + taps = make_random_float_tuple(ntaps) + expected_result = reference_filter_fff(dec, taps, src_data) + + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(dec, taps) + dst = gr.vector_sink_f() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + + self.assert_fft_float_ok2(expected_result, result_data) + + def xtest_fff_007(self): + # test decimation with nthreads + random.seed(0) + nthreads = 2 + for i in xrange(25): + sys.stderr.write("\n>>> Loop = %d\n" % (i,)) + dec = i + 1 + src_len = 4*1024 + src_data = make_random_float_tuple(src_len) + ntaps = int(random.uniform(2, 100)) + taps = make_random_float_tuple(ntaps) + expected_result = reference_filter_fff(dec, taps, src_data) + + src = gr.vector_source_f(src_data) + op = filter.fft_filter_fff(dec, taps, nthreads) + dst = gr.vector_sink_f() + tb = gr.top_block() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + + self.assert_fft_float_ok2(expected_result, result_data) + + def test_fff_get0(self): + random.seed(0) + for i in xrange(25): + ntaps = int(random.uniform(2, 100)) + taps = make_random_float_tuple(ntaps) + + op = filter.fft_filter_fff(1, taps) + result_data = op.taps() + #print result_data + + self.assertEqual(taps, result_data) + + def test_ccc_get0(self): + random.seed(0) + for i in xrange(25): + ntaps = int(random.uniform(2, 100)) + taps = make_random_complex_tuple(ntaps) + + op = filter.fft_filter_ccc(1, taps) + result_data = op.taps() + #print result_data + + self.assertComplexTuplesAlmostEqual(taps, result_data, 4) + + +if __name__ == '__main__': + gr_unittest.run(test_fft_filter, "test_fft_filter.xml") + diff --git a/gr-filter/python/qa_filter_delay_fc.py b/gr-filter/python/qa_filter_delay_fc.py new file mode 100755 index 000000000..57b063e7b --- /dev/null +++ b/gr-filter/python/qa_filter_delay_fc.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python +# +# Copyright 2004,2007,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, gr_unittest +import filter_swig as filter +import math + +class test_filter_delay_fc(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_001_filter_delay_one_input(self): + + # expected result + expected_result = ( -1.4678005338941702e-11j, + -0.0011950774351134896j, + -0.0019336787518113852j, + -0.0034673355985432863j, + -0.0036765895783901215j, + -0.004916108213365078j, + -0.0042778430506587029j, + -0.006028641015291214j, + -0.005476709920912981j, + -0.0092810001224279404j, + -0.0095402700826525688j, + -0.016060983762145042j, + -0.016446959227323532j, + -0.02523401565849781j, + -0.024382550269365311j, + -0.035477779805660248j, + -0.033021725714206696j, + -0.048487484455108643j, + -0.04543270543217659j, + -0.069477587938308716j, + -0.066984444856643677j, + -0.10703597217798233j, + -0.10620346665382385j, + -0.1852707713842392j, + -0.19357112050056458j, + (7.2191945754696007e-09 -0.50004088878631592j), + (0.58778399229049683 -0.6155126690864563j), + (0.95105588436126709 -0.12377222627401352j), + (0.95105588436126709 +0.41524654626846313j), + (0.5877838134765625 +0.91611981391906738j), + (5.8516356205018383e-09 +1.0670661926269531j), + (-0.5877840518951416 +0.87856143712997437j), + (-0.95105588436126709 +0.35447561740875244j), + (-0.95105588436126709 -0.26055556535720825j), + (-0.5877838134765625 -0.77606213092803955j), + (-8.7774534307527574e-09 -0.96460390090942383j), + (0.58778399229049683 -0.78470128774642944j), + (0.95105588436126709 -0.28380891680717468j), + (0.95105588436126709 +0.32548999786376953j), + (0.5877838134765625 +0.82514488697052002j), + (1.4629089051254596e-08 +1.0096219778060913j), + (-0.5877840518951416 +0.81836479902267456j), + (-0.95105588436126709 +0.31451958417892456j), + (-0.95105588436126709 -0.3030143678188324j), + (-0.5877838134765625 -0.80480599403381348j), + (-1.7554906861505515e-08 -0.99516552686691284j), + (0.58778399229049683 -0.80540722608566284j), + (0.95105582475662231 -0.30557557940483093j), + (0.95105588436126709 +0.31097668409347534j), + (0.5877838134765625 +0.81027895212173462j), + (2.3406542482007353e-08 +1.0000816583633423j), + (-0.5877840518951416 +0.80908381938934326j), + (-0.95105588436126709 +0.30904293060302734j), + (-0.95105588436126709 -0.30904296040534973j), + (-0.5877838134765625 -0.80908387899398804j), + (-2.6332360292258272e-08 -1.0000815391540527j), + (0.58778399229049683 -0.80908381938934326j), + (0.95105582475662231 -0.30904299020767212j), + (0.95105588436126709 +0.30904293060302734j), + (0.5877838134765625 +0.80908381938934326j), + (3.218399768911695e-08 +1.0000815391540527j)) + + tb = self.tb + + sampling_freq = 100 + + ntaps = 51 + src1 = gr.sig_source_f(sampling_freq, gr.GR_SIN_WAVE, + sampling_freq * 0.10, 1.0) + head = gr.head(gr.sizeof_float, int(ntaps + sampling_freq * 0.10)) + dst2 = gr.vector_sink_c() + + # calculate taps + taps = filter.firdes_hilbert(ntaps) + hd = filter.filter_delay_fc(taps) + + tb.connect(src1, head) + tb.connect(head, hd) + tb.connect(hd,dst2) + + tb.run() + + # get output + result_data = dst2.data() + self.assertComplexTuplesAlmostEqual(expected_result, result_data, 5) + + def test_002_filter_delay_two_inputs(self): + + # giving the same signal to both the inputs should fetch the same results + # as above + + # expected result + expected_result = ( -1.4678005338941702e-11j, + -0.0011950774351134896j, + -0.0019336787518113852j, + -0.0034673355985432863j, + -0.0036765895783901215j, + -0.004916108213365078j, + -0.0042778430506587029j, + -0.006028641015291214j, + -0.005476709920912981j, + -0.0092810001224279404j, + -0.0095402700826525688j, + -0.016060983762145042j, + -0.016446959227323532j, + -0.02523401565849781j, + -0.024382550269365311j, + -0.035477779805660248j, + -0.033021725714206696j, + -0.048487484455108643j, + -0.04543270543217659j, + -0.069477587938308716j, + -0.066984444856643677j, + -0.10703597217798233j, + -0.10620346665382385j, + -0.1852707713842392j, + -0.19357112050056458j, + (7.2191945754696007e-09 -0.50004088878631592j), + (0.58778399229049683 -0.6155126690864563j), + (0.95105588436126709 -0.12377222627401352j), + (0.95105588436126709 +0.41524654626846313j), + (0.5877838134765625 +0.91611981391906738j), + (5.8516356205018383e-09 +1.0670661926269531j), + (-0.5877840518951416 +0.87856143712997437j), + (-0.95105588436126709 +0.35447561740875244j), + (-0.95105588436126709 -0.26055556535720825j), + (-0.5877838134765625 -0.77606213092803955j), + (-8.7774534307527574e-09 -0.96460390090942383j), + (0.58778399229049683 -0.78470128774642944j), + (0.95105588436126709 -0.28380891680717468j), + (0.95105588436126709 +0.32548999786376953j), + (0.5877838134765625 +0.82514488697052002j), + (1.4629089051254596e-08 +1.0096219778060913j), + (-0.5877840518951416 +0.81836479902267456j), + (-0.95105588436126709 +0.31451958417892456j), + (-0.95105588436126709 -0.3030143678188324j), + (-0.5877838134765625 -0.80480599403381348j), + (-1.7554906861505515e-08 -0.99516552686691284j), + (0.58778399229049683 -0.80540722608566284j), + (0.95105582475662231 -0.30557557940483093j), + (0.95105588436126709 +0.31097668409347534j), + (0.5877838134765625 +0.81027895212173462j), + (2.3406542482007353e-08 +1.0000816583633423j), + (-0.5877840518951416 +0.80908381938934326j), + (-0.95105588436126709 +0.30904293060302734j), + (-0.95105588436126709 -0.30904296040534973j), + (-0.5877838134765625 -0.80908387899398804j), + (-2.6332360292258272e-08 -1.0000815391540527j), + (0.58778399229049683 -0.80908381938934326j), + (0.95105582475662231 -0.30904299020767212j), + (0.95105588436126709 +0.30904293060302734j), + (0.5877838134765625 +0.80908381938934326j), + (3.218399768911695e-08 +1.0000815391540527j)) + + + tb = self.tb + + sampling_freq = 100 + ntaps = 51 + src1 = gr.sig_source_f(sampling_freq, gr.GR_SIN_WAVE, + sampling_freq * 0.10, 1.0) + head = gr.head(gr.sizeof_float, int(ntaps + sampling_freq * 0.10)) + dst2 = gr.vector_sink_c() + + + # calculate taps + taps = filter.firdes_hilbert(ntaps) + hd = filter.filter_delay_fc(taps) + + tb.connect(src1, head) + tb.connect(head, (hd,0)) + tb.connect(head, (hd,1)) + tb.connect(hd,dst2) + tb.run() + + # get output + result_data = dst2.data() + + self.assertComplexTuplesAlmostEqual(expected_result, result_data, 5) + + + def test_003_filter_delay_two_inputs(self): + + # give two different inputs + + # expected result + expected_result = ( -0.0020331963896751404j, + -0.0016448829555884004j, + -0.0032375147566199303j, + -0.0014826074475422502j, + -0.0033034090884029865j, + -0.00051144487224519253j, + -0.0043686260469257832j, + -0.0010198024101555347j, + -0.0082517862319946289j, + -0.003456643782556057j, + -0.014193611219525337j, + -0.005875137634575367j, + -0.020293503999710083j, + -0.0067503536120057106j, + -0.026798896491527557j, + -0.0073488112539052963j, + -0.037041611969470978j, + -0.010557252913713455j, + -0.055669989436864853j, + -0.018332764506340027j, + -0.089904911816120148j, + -0.033361352980136871j, + -0.16902604699134827j, + -0.074318811297416687j, + -0.58429563045501709j, + (7.2191945754696007e-09 -0.35892376303672791j), + (0.58778399229049683 +0.63660913705825806j), + (0.95105588436126709 +0.87681591510772705j), + (0.95105588436126709 +0.98705857992172241j), + (0.5877838134765625 +0.55447429418563843j), + (5.8516356205018383e-09 +0.026006083935499191j), + (-0.5877840518951416 -0.60616838932037354j), + (-0.95105588436126709 -0.9311758279800415j), + (-0.95105588436126709 -0.96169203519821167j), + (-0.5877838134765625 -0.57292771339416504j), + (-8.7774534307527574e-09 -0.0073488391935825348j), + (0.58778399229049683 +0.59720659255981445j), + (0.95105588436126709 +0.94438445568084717j), + (0.95105588436126709 +0.95582199096679688j), + (0.5877838134765625 +0.58196049928665161j), + (1.4629089051254596e-08 +0.0026587247848510742j), + (-0.5877840518951416 -0.59129220247268677j), + (-0.95105588436126709 -0.94841635227203369j), + (-0.95105588436126709 -0.95215457677841187j), + (-0.5877838134765625 -0.58535969257354736j), + (-1.7554906861505515e-08 -0.00051158666610717773j), + (0.58778399229049683 +0.58867418766021729j), + (0.95105582475662231 +0.94965213537216187j), + (0.95105588436126709 +0.95050644874572754j), + (0.5877838134765625 +0.58619076013565063j), + (2.3406542482007353e-08 +1.1920928955078125e-07j), + (-0.5877840518951416 -0.58783555030822754j), + (-0.95105588436126709 -0.95113480091094971j), + (-0.95105588436126709 -0.95113474130630493j), + (-0.5877838134765625 -0.58783555030822754j), + (-2.6332360292258272e-08 -8.1956386566162109e-08j), + (0.58778399229049683 +0.58783555030822754j), + (0.95105582475662231 +0.95113474130630493j), + (0.95105588436126709 +0.95113474130630493j), + (0.5877838134765625 +0.58783560991287231j), + (3.218399768911695e-08 +1.1920928955078125e-07j)) + + tb = self.tb + + sampling_freq = 100 + ntaps = 51 + + src1 = gr.sig_source_f(sampling_freq, gr.GR_SIN_WAVE,sampling_freq * 0.10, 1.0) + src2 = gr.sig_source_f(sampling_freq, gr.GR_COS_WAVE,sampling_freq * 0.10, 1.0) + + head1 = gr.head(gr.sizeof_float, int(ntaps + sampling_freq * 0.10)) + head2 = gr.head(gr.sizeof_float, int(ntaps + sampling_freq * 0.10)) + + taps = filter.firdes_hilbert(ntaps) + hd = filter.filter_delay_fc(taps) + + dst2 = gr.vector_sink_c() + + tb.connect(src1, head1) + tb.connect(src2, head2) + + tb.connect(head1, (hd,0)) + tb.connect(head2, (hd,1)) + tb.connect(hd, dst2) + + tb.run() + + # get output + result_data = dst2.data() + + self.assertComplexTuplesAlmostEqual(expected_result, result_data, 5) + + +if __name__ == '__main__': + gr_unittest.run(test_filter_delay_fc, "test_filter_delay_fc.xml") diff --git a/gr-filter/python/qa_fir_filter.py b/gr-filter/python/qa_fir_filter.py new file mode 100755 index 000000000..2a61498a2 --- /dev/null +++ b/gr-filter/python/qa_fir_filter.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python +# +# Copyright 2008,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 this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +# + +from gnuradio import gr, gr_unittest +import filter_swig as filter + +class test_filter(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block () + + def tearDown(self): + self.tb = None + + def test_fir_filter_fff_001(self): + src_data = 40*[1, 2, 3, 4] + expected_data = (0.5, 1.5, 3.0, 5.0, 5.5, 6.5, 8.0, 10.0, + 10.5, 11.5, 13.0, 15.0, 15.5, 16.5, 18.0, + 20.0, 20.5, 21.5, 23.0, 25.0, 25.5, 26.5, + 28.0, 30.0, 30.5, 31.5, 33.0, 35.0, 35.5, + 36.5, 38.0, 40.0, 40.5, 41.5, 43.0, 45.0, + 45.5, 46.5, 48.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0) + src = gr.vector_source_f(src_data) + op = filter.fir_filter_fff(1, 20*[0.5, 0.5]) + dst = gr.vector_sink_f() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_fff_002(self): + src_data = 40*[1, 2, 3, 4] + expected_data = (0.5, 5.5, 10.5, 15.5, 20.5, 25.5, 30.5, 35.5, + 40.5, 45.5, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, + 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0, 50.0) + src = gr.vector_source_f(src_data) + op = filter.fir_filter_fff(4, 20*[0.5, 0.5]) + dst = gr.vector_sink_f() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_ccf_001(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((0.5+0.5j), (1.5+1.5j), (3+3j), (5+5j), (5.5+5.5j), + (6.5+6.5j), (8+8j), (10+10j), (10.5+10.5j), (11.5+11.5j), + (13+13j), (15+15j), (15.5+15.5j), (16.5+16.5j), (18+18j), + (20+20j), (20.5+20.5j), (21.5+21.5j), (23+23j), (25+25j), + (25.5+25.5j), (26.5+26.5j), (28+28j), (30+30j), (30.5+30.5j), + (31.5+31.5j), (33+33j), (35+35j), (35.5+35.5j), (36.5+36.5j), + (38+38j), (40+40j), (40.5+40.5j), (41.5+41.5j), (43+43j), + (45+45j), (45.5+45.5j), (46.5+46.5j), (48+48j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j)) + + src = gr.vector_source_c(src_data) + op = filter.fir_filter_ccf(1, 20*[0.5, 0.5]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_ccf_002(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((0.5+0.5j), (5.5+5.5j), (10.5+10.5j), (15.5+15.5j), + (20.5+20.5j), (25.5+25.5j), (30.5+30.5j), (35.5+35.5j), + (40.5+40.5j), (45.5+45.5j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j), (50+50j), + (50+50j), (50+50j), (50+50j), (50+50j)) + + src = gr.vector_source_c(src_data) + op = filter.fir_filter_ccf(4, 20*[0.5, 0.5]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_ccc_001(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((-0.5+1.5j), (-1.5+4.5j), (-3+9j), (-5+15j), + (-5.5+16.5j), (-6.5+19.5j), (-8+24j), (-10+30j), + (-10.5+31.5j), (-11.5+34.5j), (-13+39j), (-15+45j), + (-15.5+46.5j), (-16.5+49.5j), (-18+54j), (-20+60j), + (-20.5+61.5j), (-21.5+64.5j), (-23+69j), (-25+75j), + (-25.5+76.5j), (-26.5+79.5j), (-28+84j), (-30+90j), + (-30.5+91.5j), (-31.5+94.5j), (-33+99j), (-35+105j), + (-35.5+106.5j), (-36.5+109.5j), (-38+114j), (-40+120j), + (-40.5+121.5j), (-41.5+124.5j), (-43+129j), (-45+135j), + (-45.5+136.5j), (-46.5+139.5j), (-48+144j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), (-50+150j)) + src = gr.vector_source_c(src_data) + op = filter.fir_filter_ccc(1, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + + def test_fir_filter_ccc_002(self): + src_data = 10*[1+1j, 2+2j, 3+3j, 4+4j] + + # results derived from original gr.fir_filter_ccc + expected_data = ((7.537424837948042e-20+7.537424837948042e-20j), (9.131923434324563e-05+9.131923434324563e-05j), (0.0003317668742965907+0.0003317668742965907j), (0.0007230418268591166+0.0007230418268591166j), (0.0012087896466255188+0.0012087896466255188j), (0.0013292605290189385+0.0013292605290189385j), (0.001120875240303576+0.001120875240303576j), (0.000744672492146492+0.000744672492146492j), (0.000429437990533188+0.000429437990533188j), (2.283908543176949e-05+2.283908543176949e-05j), (-0.0002245186478830874-0.0002245186478830874j), (-0.0001157080550910905-0.0001157080550910905j), (0.00041409023106098175+0.00041409023106098175j), (0.0009017843985930085+0.0009017843985930085j), (0.0012520025484263897+0.0012520025484263897j), (0.0014116164529696107+0.0014116164529696107j), (0.001393353333696723+0.001393353333696723j), (0.000912194955162704+0.000912194955162704j), (0.00022649182938039303+0.00022649182938039303j), (-0.00031363096786662936-0.00031363096786662936j), (-0.0003966730728279799-0.0003966730728279799j), (-0.00023757052258588374-0.00023757052258588374j), (0.00021952332463115454+0.00021952332463115454j), (0.0009092430118471384+0.0009092430118471384j), (0.001662317430600524+0.001662317430600524j), (0.0019024648936465383+0.0019024648936465383j), (0.0015955769922584295+0.0015955769922584295j), (0.0009144138311967254+0.0009144138311967254j), (0.0001872836146503687+0.0001872836146503687j), (-0.000581968342885375-0.000581968342885375j), (-0.0009886166080832481-0.0009886166080832481j), (-0.0007480768254026771-0.0007480768254026771j), (0.00018211957649327815+0.00018211957649327815j), (0.0012042406015098095+0.0012042406015098095j), (0.0020200139842927456+0.0020200139842927456j), (0.0023816542234271765+0.0023816542234271765j), (0.002195809967815876+0.002195809967815876j), (0.0012113333214074373+0.0012113333214074373j), (-0.00014088614261709154-0.00014088614261709154j), (-0.0012574587017297745-0.0012574587017297745j)) + + taps = filter.firdes.low_pass(1, 1, 0.1, 0.01) + src = gr.vector_source_c(src_data) + op = filter.fir_filter_ccc(1, taps) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_ccc_003(self): + src_data = 40*[1+1j, 2+2j, 3+3j, 4+4j] + expected_data = ((-0.5+1.5j), (-5.5+16.5j), (-10.5+31.5j), + (-15.5+46.5j), (-20.5+61.5j), (-25.5+76.5j), + (-30.5+91.5j), (-35.5+106.5j), (-40.5+121.5j), + (-45.5+136.5j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j), (-50+150j), + (-50+150j), (-50+150j), (-50+150j)) + src = gr.vector_source_c(src_data) + op = filter.fir_filter_ccc(4, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + + def test_fir_filter_scc_001(self): + src_data = 40*[1, 2, 3, 4] + expected_data = ((0.5+1j), (1.5+3j), (3+6j), (5+10j), (5.5+11j), + (6.5+13j), (8+16j), (10+20j), (10.5+21j), (11.5+23j), + (13+26j), (15+30j), (15.5+31j), (16.5+33j), (18+36j), + (20+40j), (20.5+41j), (21.5+43j), (23+46j), (25+50j), + (25.5+51j), (26.5+53j), (28+56j), (30+60j), (30.5+61j), + (31.5+63j), (33+66j), (35+70j), (35.5+71j), (36.5+73j), + (38+76j), (40+80j), (40.5+81j), (41.5+83j), (43+86j), + (45+90j), (45.5+91j), (46.5+93j), (48+96j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j)) + src = gr.vector_source_s(src_data) + op = filter.fir_filter_scc(1, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + + def test_fir_filter_scc_002(self): + src_data = 40*[1, 2, 3, 4] + expected_data = ((0.5+1j), (5.5+11j), (10.5+21j), (15.5+31j), (20.5+41j), + (25.5+51j), (30.5+61j), (35.5+71j), (40.5+81j), (45.5+91j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j), + (50+100j), (50+100j), (50+100j), (50+100j), (50+100j)) + src = gr.vector_source_s(src_data) + op = filter.fir_filter_scc(4, 20*[0.5+1j, 0.5+1j]) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + def test_fir_filter_fsf_001(self): + src_data = 40*[1, 2, 3, 4] + expected_data =(0, 1, 3, 5, 5, 6, 8, 10, 10, 11, 13, 15, 15, 16, 18, 20, 20, + 21, 23, 25, 25, 26, 28, 30, 30, 31, 33, 35, 35, 36, 38, 40, 40, + 41, 43, 45, 45, 46, 48, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50) + src = gr.vector_source_f(src_data) + op = filter.fir_filter_fsf(1, 20*[0.5, 0.5]) + dst = gr.vector_sink_s() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + + + def test_fir_filter_fsf_002(self): + src_data = 40*[1, 2, 3, 4] + expected_data = (0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, + 50, 50, 50, 50, 50, 50, 50, 50, 50, 50) + src = gr.vector_source_f(src_data) + op = filter.fir_filter_fsf(4, 20*[0.5, 0.5]) + dst = gr.vector_sink_s() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data, 5) + +if __name__ == '__main__': + gr_unittest.run(test_filter, "test_filter.xml") + diff --git a/gr-filter/python/qa_firdes.py b/gr-filter/python/qa_firdes.py new file mode 100755 index 000000000..cfd10435f --- /dev/null +++ b/gr-filter/python/qa_firdes.py @@ -0,0 +1,202 @@ +#!/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 sys + +class test_firdes(gr_unittest.TestCase): + + def setUp(self): + pass + + def tearDown(self): + pass + + def test_low_pass(self): + known_taps = (0.0030193300917744637, -0.004960992839187384, + 0.006678304169327021, -1.132049690556083e-17, + -0.0251916591078043, 0.07206480950117111, + -0.13062666356563568, 0.18007083237171173, + 0.7978920936584473, 0.18007083237171173, + -0.13062666356563568, 0.07206480950117111, + -0.0251916591078043, -1.132049690556083e-17, + 0.006678304169327021, -0.004960992839187384, + 0.0030193300917744637) + new_taps = filter.firdes.low_pass(1, 1, 0.4, 0.2) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_low_pass_2(self): + known_taps = (0.0024871660862118006, -4.403502608370943e-18, + -0.014456653036177158, 0.0543283149600029, + -0.116202212870121, 0.17504146695137024, + 0.7976038455963135, 0.17504146695137024, + -0.116202212870121, 0.0543283149600029, + -0.014456653036177158, -4.403502608370943e-18, + 0.0024871660862118006) + new_taps = filter.firdes.low_pass_2(1, 1, 0.4, 0.2, 60) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_high_pass(self): + known_taps = (-0.003062003292143345, 0.005031108390539885, + -0.0067726909182965755, 1.1480492661182674e-17, + 0.025547700002789497, -0.0730833187699318, + 0.13247284293174744, -0.18261581659317017, + 0.20229223370552063, -0.18261581659317017, + 0.13247284293174744, -0.0730833187699318, + 0.025547700002789497, 1.1480492661182674e-17, + -0.0067726909182965755, 0.005031108390539885, + -0.003062003292143345) + new_taps = filter.firdes.high_pass(1, 1, 0.4, 0.2) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_high_pass_2(self): + known_taps = (-0.0027197482995688915, 4.815287179370254e-18, + 0.01580853760242462, -0.05940871313214302, + 0.1270686239004135, -0.1914101094007492, + 0.21804752945899963, -0.1914101094007492, + 0.1270686239004135, -0.05940871313214302, + 0.01580853760242462, 4.815287179370254e-18, + -0.0027197482995688915) + new_taps = filter.firdes.high_pass_2(1, 1, 0.4, 0.2, 60) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_band_pass(self): + known_taps = (0.004961997736245394, -0.008152946829795837, + -0.004192151129245758, -5.749020235348687e-18, + 0.01581347920000553, 0.11843203753232956, + -0.21467317640781403, -0.11303528398275375, + 0.40520283579826355, -0.11303528398275375, + -0.21467317640781403, 0.11843203753232956, + 0.01581347920000553, -5.749020235348687e-18, + -0.004192151129245758, -0.008152946829795837, + 0.004961997736245394) + new_taps = filter.firdes.band_pass(1, 1, 0.2, 0.4, 0.2) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_band_pass_2(self): + known_taps = (-0.001676854444667697, -2.4018533253972557e-18, + 0.009746716357767582, 0.09589414298534393, + -0.20510689914226532, -0.11801345646381378, + 0.4350462853908539, -0.11801345646381378, + -0.20510689914226532, 0.09589414298534393, + 0.009746716357767582, -2.4018533253972557e-18, + -0.001676854444667697) + new_taps = filter.firdes.band_pass_2(1, 1, 0.2, 0.4, 0.2, 60) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_complex_band_pass(self): + known_taps = ((0.0024772135075181723+0.0017997993854805827j), + (-0.004070250317454338+0.002957213670015335j), + (-0.0020928815938532352-0.006441210396587849j), + (-2.8701231652956686e-18+2.805614574993832e-24j), + (0.007894645445048809-0.024297315627336502j), + (0.05912570655345917+0.04295721650123596j), + (-0.10717268288135529+0.07786571979522705j), + (-0.0564316064119339-0.17367789149284363j), + (0.20229223370552063-2.4115112751132983e-07j), + (-0.05643119290471077+0.17367802560329437j), + (-0.10717286914587021-0.07786546647548676j), + (0.05912560224533081-0.0429573580622673j), + (0.007894691079854965+0.024297300726175308j), + (-2.8701231652956686e-18+2.6687109203363464e-24j), + (-0.0020928694866597652+0.006441214121878147j), + (-0.004070255905389786-0.0029572059866040945j), + (0.0024772100150585175-0.0017998040420934558j)) + new_taps = filter.firdes.complex_band_pass(1, 1, 0.2, 0.4, 0.2) + self.assertComplexTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_complex_band_pass_2(self): + known_taps = ((-0.0008404505206272006-0.0025866336654871702j), + (-1.2038217948425635e-18+1.1767648157397848e-24j), + (0.0048850891180336475-0.015034818090498447j), + (0.048062704503536224+0.03491950035095215j), + (-0.10280057787895203+0.07468919456005096j), + (-0.05914920195937157-0.18204176425933838j), + (0.21804752945899963-2.5993290364567656e-07j), + (-0.059148769825696945+0.18204189836978912j), + (-0.10280075669288635-0.07468894869089127j), + (0.04806262254714966-0.0349196158349514j), + (0.004885117989033461+0.015034808777272701j), + (-1.2038217948425635e-18+1.1193430388030685e-24j), + (-0.000840445572976023+0.002586635295301676j)) + new_taps = filter.firdes.complex_band_pass_2(1, 1, 0.2, 0.4, 0.2, 60) + self.assertComplexTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_band_reject(self): + known_taps = (-0.004915320314466953, 0.008076251484453678, + 0.00415271520614624, 5.694938753309664e-18, + -0.01566472090780735, -0.11731793731451035, + 0.2126537412405014, 0.11197195947170258, + 0.6020866632461548, 0.11197195947170258, + 0.2126537412405014, -0.11731793731451035, + -0.01566472090780735, 5.694938753309664e-18, + 0.00415271520614624, 0.008076251484453678, + -0.004915320314466953) + new_taps = filter.firdes.band_reject(1, 1, 0.2, 0.4, 0.2) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_band_reject_2(self): + known_taps = (0.0015371545450761914, 2.201753372137003e-18, + -0.00893471110612154, -0.08790513873100281, + 0.1880193054676056, 0.1081816703081131, + 0.5982034206390381, 0.1081816703081131, + 0.1880193054676056, -0.08790513873100281, + -0.00893471110612154, 2.201753372137003e-18, + 0.0015371545450761914) + new_taps = filter.firdes.band_reject_2(1, 1, 0.2, 0.4, 0.2, 60) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_hilbert(self): + known_taps = (-0.010056184604763985, 0.0, + -0.08335155993700027, 0.0, + -0.5732954144477844, 0.0, + 0.5732954144477844, 0.0, + 0.08335155993700027, 0.0, + 0.010056184604763985) + new_taps = filter.firdes.hilbert(11) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_root_raised_cosine(self): + known_taps = (-0.04609205573797226, -0.02069387212395668, + 0.050548505038022995, 0.14850808680057526, + 0.23387153446674347, 0.2677156329154968, + 0.23387153446674347, 0.14850808680057526, + 0.050548505038022995, -0.02069387212395668, + -0.04609205573797226) + new_taps = filter.firdes.root_raised_cosine(1, 4, 1, 0.35, 11) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + + def test_gaussian(self): + known_taps = (0.0003600157215259969, 0.0031858310103416443, + 0.0182281993329525, 0.06743486225605011, + 0.16130395233631134, 0.24947398900985718, + 0.24947398900985718, 0.16130395233631134, + 0.06743486225605011, 0.0182281993329525, + 0.0031858310103416443, 0.0003600157215259969, + 2.630509879963938e-05) + new_taps = filter.firdes.gaussian(1, 4, 0.35, 13) + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + +if __name__ == '__main__': + gr_unittest.run(test_firdes, "test_firdes.xml") + diff --git a/gr-filter/python/qa_fractional_interpolator.py b/gr-filter/python/qa_fractional_interpolator.py new file mode 100755 index 000000000..9e0f685d8 --- /dev/null +++ b/gr-filter/python/qa_fractional_interpolator.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python +# +# Copyright 2007,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, gr_unittest +import filter_swig as filter +import math + +class test_fractional_resampler(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_001_ff(self): + N = 10000 # number of samples to use + fs = 1000 # baseband sampling rate + rrate = 1.123 # resampling rate + + freq = 10 + signal = gr.sig_source_f(fs, gr.GR_SIN_WAVE, freq, 1) + head = gr.head(gr.sizeof_float, N) + op = filter.fractional_interpolator_ff(0, rrate) + snk = gr.vector_sink_f() + + self.tb.connect(signal, head, op, snk) + self.tb.run() + + Ntest = 5000 + L = len(snk.data()) + t = map(lambda x: float(x)/(fs/rrate), xrange(L)) + + phase = 0.1884 + expected_data = map(lambda x: math.sin(2.*math.pi*freq*x+phase), t) + + dst_data = snk.data() + + self.assertFloatTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 3) + + + def test_002_cc(self): + N = 10000 # number of samples to use + fs = 1000 # baseband sampling rate + rrate = 1.123 # resampling rate + + freq = 10 + signal = gr.sig_source_c(fs, gr.GR_SIN_WAVE, freq, 1) + head = gr.head(gr.sizeof_gr_complex, N) + op = filter.fractional_interpolator_cc(0.0, rrate) + snk = gr.vector_sink_c() + + self.tb.connect(signal, head, op, snk) + self.tb.run() + + Ntest = 5000 + L = len(snk.data()) + t = map(lambda x: float(x)/(fs/rrate), xrange(L)) + + phase = 0.1884 + expected_data = map(lambda x: math.cos(2.*math.pi*freq*x+phase) + \ + 1j*math.sin(2.*math.pi*freq*x+phase), t) + + dst_data = snk.data() + + self.assertComplexTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 3) + + +if __name__ == '__main__': + gr_unittest.run(test_fractional_resampler, "test_fractional_resampler.xml") diff --git a/gr-filter/python/qa_freq_xlating_fir_filter.py b/gr-filter/python/qa_freq_xlating_fir_filter.py new file mode 100755 index 000000000..ee38eb7df --- /dev/null +++ b/gr-filter/python/qa_freq_xlating_fir_filter.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python +# +# Copyright 2008,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 this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +# + +from gnuradio import gr, gr_unittest +import filter_swig as filter +import cmath, math + +class test_freq_xlating_filter(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block () + + def tearDown(self): + self.tb = None + + def generate_ccf_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.low_pass(1, fs, bw, bw/4) + times = xrange(100) + self.src_data = map(lambda t: cmath.exp(-2j*cmath.pi*fc/fs*(t/100.0)), times) + + def generate_ccc_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.complex_band_pass(1, fs, -bw/2, bw/2, bw/4) + times = xrange(100) + self.src_data = map(lambda t: cmath.exp(-2j*cmath.pi*fc/fs*(t/100.0)), times) + + def generate_fcf_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.low_pass(1, fs, bw, bw/4) + times = xrange(100) + self.src_data = map(lambda t: math.sin(2*cmath.pi*fc/fs*(t/100.0)), times) + + def generate_fcc_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.complex_band_pass(1, fs, -bw/2, bw/2, bw/4) + times = xrange(100) + self.src_data = map(lambda t: math.sin(2*cmath.pi*fc/fs*(t/100.0)), times) + + def generate_scf_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.low_pass(1, fs, bw, bw/4) + times = xrange(100) + self.src_data = map(lambda t: int(100*math.sin(2*cmath.pi*fc/fs*(t/100.0))), times) + + def generate_scc_source(self): + self.fs = fs = 1 + self.fc = fc = 0.3 + self.bw = bw = 0.1 + self.taps = filter.firdes.complex_band_pass(1, fs, -bw/2, bw/2, bw/4) + times = xrange(100) + self.src_data = map(lambda t: int(100*math.sin(2*cmath.pi*fc/fs*(t/100.0))), times) + + + def test_fir_filter_ccf_001(self): + self.generate_ccf_source() + expected_data = ((0.001697700354270637+0.004312471952289343j), + (0.003520616563037038-0.003014103975147009j), + (0.004252811893820763-0.008337559178471565j), + (0.0030743128154426813-0.010262271389365196j), + (0.0007344777695834637-0.007861139252781868j), + (-0.0011067686136811972-0.0028924935031682253j), + (-0.002371778478845954+0.0019914964213967323j), + (-0.003023319412022829+0.005717850290238857j), + (-0.0021738125942647457+0.007211698684841394j), + (-0.0004628606839105487+0.005501383915543556j), + (0.0007428556564264+0.0019867848604917526j), + (0.001634795218706131-0.0013514887541532516j), + (0.002205110155045986-0.00402155052870512j), + (0.0015480631263926625-0.005179159343242645j), + (0.00026722141774371266-0.003887997241690755j), + (-0.0004911854630336165-0.0013578246580436826j), + (-0.0011226939968764782+0.0009080552263185382j), + (-0.0016229727771133184+0.0028335191309452057j), + (-0.0010890064295381308+0.0037298379465937614j), + (-0.00012392725329846144+0.0027196139562875032j)) + src = gr.vector_source_c(self.src_data) + op = filter.freq_xlating_fir_filter_ccf(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_ccf_002(self): + self.generate_ccf_source() + expected_data = ((6.419439159799367e-05-0.0006292851758189499j), + (-0.00037074743886478245+0.0013245552545413375j), + (0.0006853155209682882-0.0023769831750541925j), + (-0.001427714480087161+0.002608160488307476j), + (0.0015907397028058767-0.000811046629678458j), + (-0.0004226673918310553-0.0024389736354351044j), + (-0.0013841050677001476+0.006231029983609915j), + (0.0035029184073209763-0.009738259017467499j), + (-0.005924836732447147+0.010320881381630898j), + (0.006831614300608635-0.003950652200728655j), + (-0.0021247887052595615-0.015604906715452671j), + (-0.04283163696527481+0.09995654970407486j), + (-0.01391829177737236+0.07924056798219681j), + (0.010886997915804386-0.02463012933731079j), + (-0.0056075905449688435+0.004998659715056419j), + (0.0016976913902908564+0.004312459379434586j), + (0.0007344821933656931-0.007861112244427204j), + (-0.002173811662942171+0.007211671676486731j), + (0.0022051059640944004-0.00402153329923749j), + (-0.0011226903880015016+0.0009080505697056651j)) + src = gr.vector_source_c(self.src_data) + op = filter.freq_xlating_fir_filter_ccf(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_ccc_001(self): + self.generate_ccc_source() + expected_data = ((0.0036842757836-0.0114002721384j), + (0.00324621866457-0.0108166672289j), + (0.00206564785913-0.00923090614378j), + (0.00109899020754-0.00656201224774j), + (0.000506619049702-0.00402844604105j), + (-0.000523390364833-0.00166808743961j), + (-0.00140534969978+0.00103991874494j), + (-0.00154365820345+0.00315759982914j), + (-0.00180402118713+0.00427215453237j), + (-0.00216706306674+0.00524478312582j), + (-0.00178848754149+0.0057489364408j), + (-0.00129876169376+0.00512680830434j), + (-0.00122803379782+0.00427244976163j), + (-0.000722666736692+0.00351428100839j), + (5.53092104383e-05+0.00207865727134j), + (0.000227351076319+0.000517217209563j), + (0.000414477253798-0.000383921898901j), + (0.000998671515845-0.00135387131013j), + (0.00104933069088-0.00243046949618j), + (0.000765930046327-0.0026717747096j)) + src = gr.vector_source_c(self.src_data) + op = filter.freq_xlating_fir_filter_ccc(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_ccc_002(self): + self.generate_ccc_source() + expected_data = ((-0.000650451984257+0.00120697380044j), + (-9.59713361226e-05+0.00102412770502j), + (0.000958710326813-0.00145424995571j), + (0.000901343999431-0.00290832063183j), + (-0.000822560978122+0.000296717538731j), + (-0.00211223773658+0.00519825471565j), + (-0.00037001183955+0.00358242215589j), + (0.00327983591706-0.00616005761549j), + (0.00356886954978-0.0117237549275j), + (-0.00328874029219+0.00182871113066j), + (-0.0139285130426+0.0320657044649j), + (-0.0198133718222+0.0562113076448j), + (-0.0157803222537+0.0530290603638j), + (-0.00550725404173+0.0255754813552j), + (0.00252919178456-0.00232240976766j), + (0.00368427345529-0.0114002330229j), + (0.000506620621309-0.00402843113989j), + (-0.00180401885882+0.00427213776857j), + (-0.00122803344857+0.00427243299782j), + (0.000414476031438-0.000383919978049j)) + src = gr.vector_source_c(self.src_data) + op = filter.freq_xlating_fir_filter_ccc(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_fcf_001(self): + self.generate_fcf_source() + expected_data = ((0.000247188087087-0.000157509770361j), + (-0.000155229790835-0.000246197130764j), + (-0.000264906557277+0.000174603672349j), + (6.99016964063e-05+0.000174961372977j), + (-5.48477692064e-05-0.0001131295503j), + (-0.000237467131228+0.000118011368613j), + (0.000136614587973+0.000229531884543j), + (0.000229347482673-0.000166581812664j), + (-0.000106010869786-0.000150042149471j), + (2.92293734674e-05+0.000142060467624j), + (0.000228707227507-9.30760797928e-05j), + (-0.000124306126963-0.000216641055886j), + (-0.000204823678359+0.00016052465071j), + (0.00012825592421+0.000133123627165j), + (-1.18284006021e-05-0.000159015646204j), + (-0.000219973371713+7.5438656495e-05j), + (0.000114713984658+0.000205190401175j), + (0.000185727752978-0.000154630601173j), + (-0.000141745767905-0.000120098840853j), + (-3.9850056055e-07+0.000168364742422j)) + src = gr.vector_source_f(self.src_data) + op = filter.freq_xlating_fir_filter_fcf(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_fcf_002(self): + self.generate_fcf_source() + expected_data = ((7.3052920925e-05-3.91741014028e-06j), + (3.11913172482e-05-0.000109872074972j), + (-0.000128646017401-3.49857727997e-05j), + (-5.49546712136e-05+8.96326746442e-05j), + (5.14321582159e-05+9.64698920143e-05j), + (0.000120189361041+2.4231892894e-05j), + (0.000100405508419-0.000223224604269j), + (-0.000274751859251-2.33274622587e-05j), + (1.52600114234e-06+0.000133301247843j), + (3.77224641852e-05+5.29596509296e-05j), + (-3.60160379387e-06+0.000247975171078j), + (0.00113093166146-0.000663110695314j), + (0.00059568521101-0.00099650840275j), + (-0.000475480686873+0.000250602373853j), + (0.000191397906747+0.000271986238658j), + (0.000247183139436-0.000157510468853j), + (-5.48357638763e-05-0.000113135029096j), + (-0.00010601492977-0.00015005269961j), + (-0.000204817260965+0.000160534662427j), + (0.000114742244477+0.000205190313864j)) + src = gr.vector_source_f(self.src_data) + op = filter.freq_xlating_fir_filter_fcf(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_fcc_001(self): + self.generate_fcc_source() + expected_data = ((-0.000337088305969+7.46004516259e-05j), + (-5.63409266761e-05+0.000301144464174j), + (9.16960561881e-05-2.89259278361e-05j), + (-0.000231866899412-6.25764005235e-05j), + (-0.000116414688819+0.000258557556663j), + (0.000206079319469+5.05045172758e-05j), + (-3.85114690289e-05-0.00019276549574j), + (-0.000146380873048+0.000112079876999j), + (0.000215423395275+0.000116706331028j), + (0.000136050162837-0.000232611957472j), + (-0.000155499437824-5.41604022146e-05j), + (0.000106907449663+0.00016310159117j), + (0.000224392410018-0.000156331108883j), + (-0.000131131906528-0.000172063446371j), + (-5.92393880652e-05+0.00016801241145j), + (0.000214921761653-5.32235890205e-06j), + (-5.96960526309e-05-0.000211164733628j), + (-0.000193948610104+0.000113364716526j), + (0.000134820176754+0.000142527525895j), + (4.74465123261e-05-0.000175131688593j)) + src = gr.vector_source_f(self.src_data) + op = filter.freq_xlating_fir_filter_fcc(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_fcc_002(self): + self.generate_fcc_source() + expected_data = ((-6.94218761055e-05-9.90489479591e-06j), + (-2.56973435171e-05+8.05932795629e-05j), + (7.01698663761e-05+7.36373840482e-05j), + (7.57163215894e-05-4.65324592369e-05j), + (-3.01657128148e-05-0.000122838056996j), + (-9.53659764491e-05-3.73945695173e-05j), + (-2.33501577895e-05+0.000109135726234j), + (6.65136758471e-05+0.000125709688291j), + (3.08501912514e-05-9.16842873266e-06j), + (-2.64703612629e-05-0.000135892929393j), + (0.000136643866426-0.000162003751029j), + (0.000501801609062-0.000185820827028j), + (0.000694551155902-0.000299874518532j), + (0.000424396857852-0.00038379128091j), + (-9.1786707344e-05-0.000242479465669j), + (-0.000337087287335+7.45999423089e-05j), + (-0.000116414521472+0.000258556567132j), + (0.000215422536712+0.000116706112749j), + (0.000224391726078-0.000156330308528j), + (-5.96956087975e-05-0.000211163976928j)) + src = gr.vector_source_f(self.src_data) + op = filter.freq_xlating_fir_filter_fcc(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_scf_001(self): + self.generate_scf_source() + expected_data = ((-0.0330070219934+0.101965591311j), + (-0.0484239049256+0.0872343629599j), + (-0.0214109234512+0.102555386722j), + (0.0484058149159+0.0557125210762j), + (0.0574690811336-0.0350844524801j), + (0.0365394353867-0.0802438184619j), + (0.0453781634569-0.130992040038j), + (0.00801951438189-0.214278846979j), + (-0.0770946145058-0.230616629124j), + (-0.105601429939-0.190731987357j), + (-0.105361394584-0.177761554718j), + (-0.131518915296-0.136102750897j), + (-0.103761836886-0.0382263250649j), + (-0.0167790111154+0.0152206514031j), + (0.0277570039034+0.0300403907895j), + (0.056065287441+0.0806603953242j), + (0.118084669113+0.104863211513j), + (0.128281414509+0.0677760615945j), + (0.0748447552323+0.0619902014732j), + (0.0512856245041+0.0775099247694j)) + src = gr.vector_source_s(self.src_data) + op = filter.freq_xlating_fir_filter_scf(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_scf_002(self): + self.generate_scf_source() + expected_data = ((0.00824625696987-1.50158575707e-05j), + (0.0020101047121-0.0116540221497j), + (-0.0126378880814-0.00259830290452j), + (-0.00363933457993+0.00837504956871j), + (0.00107023562305+0.00915473792702j), + (0.0169738996774+0.00422182958573j), + (0.00630031805485-0.025423232466j), + (-0.0283014029264+0.00104465708137j), + (0.00890890974551+0.0115978596732j), + (-0.0142687577754+0.00306978379376j), + (0.02845691517+0.0331163145602j), + (0.0538152232766-0.0908300876617j), + (-0.0843691527843-0.0956566259265j), + (0.0476895272732+0.0747984498739j), + (0.0898786485195+0.082478672266j), + (-0.0330070182681+0.101965606213j), + (0.0574697069824-0.0350842289627j), + (-0.0770940706134-0.230615705252j), + (-0.103762261569-0.0382265634835j), + (0.11808334291+0.104863762856j)) + src = gr.vector_source_s(self.src_data) + op = filter.freq_xlating_fir_filter_scf(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_scc_001(self): + self.generate_scc_source() + expected_data = ((-0.00775564694777+0.0437113791704j), + (0.0108830630779+0.0433648750186j), + (0.015553932637-0.0133284125477j), + (-0.0264905355871-0.0403266139328j), + (-0.0243480335921-0.03030154109j), + (-0.000327925226884-0.069333948195j), + (-0.0306392069906-0.107313856483j), + (-0.0452371090651-0.0854917764664j), + (-0.0108894333243-0.0875641107559j), + (-0.0182112380862-0.118961036205j), + (-0.0447825863957-0.0922874584794j), + (-0.0147479763255-0.0572904124856j), + (0.00204290449619-0.0721436738968j), + (-0.027713002637-0.0548989400268j), + (-0.0149045493454-0.00210141134448j), + (0.0176361314952-0.00149522523861j), + (-0.00527482619509-0.00698099425063j), + (-0.0151527002454+0.036265052855j), + (0.0199296213686+0.0452499426901j), + (0.0122985243797+0.0143278446048j)) + src = gr.vector_source_s(self.src_data) + op = filter.freq_xlating_fir_filter_scc(1, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + + def test_fir_filter_scc_002(self): + self.generate_scc_source() + expected_data = ((-0.0080680437386-0.00158522999845j), + (-0.0037795654498+0.00733159901574j), + (0.00842926371843+0.00777021236718j), + (0.0112090632319-0.00249325321056j), + (-0.0027476802934-0.0115710813552j), + (-0.0158688724041-0.00715934624895j), + (-0.00802888441831+0.00620818417519j), + (0.0131985172629+0.0149175003171j), + (0.0190298333764+0.00956719089299j), + (-0.00112380902283-0.00936658866704j), + (-0.0204226914793-0.0333464704454j), + (-0.00695514678955-0.0437445007265j), + (0.0314490310848-0.0207983348519j), + (0.0529675260186+0.0302227605134j), + (0.0317338332534+0.0667510479689j), + (-0.00775565672666+0.0437112376094j), + (-0.024347923696-0.0303014591336j), + (-0.0108893103898-0.0875638127327j), + (0.00204296782613-0.0721434056759j), + (-0.00527479872108-0.00698097236454j)) + src = gr.vector_source_s(self.src_data) + op = filter.freq_xlating_fir_filter_scc(4, self.taps, self.fc, self.fs) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_data, result_data[-20:], 5) + +if __name__ == '__main__': + gr_unittest.run(test_freq_xlating_filter, "test_freq_xlating_filter.xml") + diff --git a/gr-filter/python/qa_hilbert.py b/gr-filter/python/qa_hilbert.py new file mode 100755 index 000000000..b460b6438 --- /dev/null +++ b/gr-filter/python/qa_hilbert.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python +# +# Copyright 2004,2007,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, gr_unittest +import filter_swig as filter +import math + +class test_hilbert(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block () + + def tearDown(self): + self.tb = None + + def test_hilbert(self): + tb = self.tb + ntaps = 51 + sampling_freq = 100 + + expected_result = ( -1.4678005338941702e-11j, + -0.0011950774351134896j, + -0.0019336787518113852j, + -0.0034673355985432863j, + -0.0036765895783901215j, + -0.004916108213365078j, + -0.0042778430506587029j, + -0.006028641015291214j, + -0.005476709920912981j, + -0.0092810001224279404j, + -0.0095402700826525688j, + -0.016060983762145042j, + -0.016446959227323532j, + -0.02523401565849781j, + -0.024382550269365311j, + -0.035477779805660248j, + -0.033021725714206696j, + -0.048487484455108643j, + -0.04543270543217659j, + -0.069477587938308716j, + -0.066984444856643677j, + -0.10703597217798233j, + -0.10620346665382385j, + -0.1852707713842392j, + -0.19357112050056458j, + (7.2191945754696007e-09 -0.50004088878631592j), + (0.58778399229049683 -0.6155126690864563j), + (0.95105588436126709 -0.12377222627401352j), + (0.95105588436126709 +0.41524654626846313j), + (0.5877838134765625 +0.91611981391906738j), + (5.8516356205018383e-09 +1.0670661926269531j), + (-0.5877840518951416 +0.87856143712997437j), + (-0.95105588436126709 +0.35447561740875244j), + (-0.95105588436126709 -0.26055556535720825j), + (-0.5877838134765625 -0.77606213092803955j), + (-8.7774534307527574e-09 -0.96460390090942383j), + (0.58778399229049683 -0.78470128774642944j), + (0.95105588436126709 -0.28380891680717468j), + (0.95105588436126709 +0.32548999786376953j), + (0.5877838134765625 +0.82514488697052002j), + (1.4629089051254596e-08 +1.0096219778060913j), + (-0.5877840518951416 +0.81836479902267456j), + (-0.95105588436126709 +0.31451958417892456j), + (-0.95105588436126709 -0.3030143678188324j), + (-0.5877838134765625 -0.80480599403381348j), + (-1.7554906861505515e-08 -0.99516552686691284j), + (0.58778399229049683 -0.80540722608566284j), + (0.95105582475662231 -0.30557557940483093j), + (0.95105588436126709 +0.31097668409347534j), + (0.5877838134765625 +0.81027895212173462j), + (2.3406542482007353e-08 +1.0000816583633423j), + (-0.5877840518951416 +0.80908381938934326j), + (-0.95105588436126709 +0.30904293060302734j), + (-0.95105588436126709 -0.30904296040534973j), + (-0.5877838134765625 -0.80908387899398804j), + (-2.6332360292258272e-08 -1.0000815391540527j), + (0.58778399229049683 -0.80908381938934326j), + (0.95105582475662231 -0.30904299020767212j), + (0.95105588436126709 +0.30904293060302734j), + (0.5877838134765625 +0.80908381938934326j), + (3.218399768911695e-08 +1.0000815391540527j)) + + + src1 = gr.sig_source_f(sampling_freq, gr.GR_SIN_WAVE, + sampling_freq * 0.10, 1.0) + + head = gr.head(gr.sizeof_float, int (ntaps + sampling_freq * 0.10)) + hilb = filter.hilbert_fc(ntaps) + dst1 = gr.vector_sink_c() + tb.connect(src1, head) + tb.connect(head, hilb) + tb.connect(hilb, dst1) + tb.run() + dst_data = dst1.data() + self.assertComplexTuplesAlmostEqual (expected_result, dst_data, 5) + +if __name__ == '__main__': + gr_unittest.run(test_hilbert, "test_hilbert.xml") diff --git a/gr-filter/python/qa_iir_filter.py b/gr-filter/python/qa_iir_filter.py new file mode 100755 index 000000000..645c4b66e --- /dev/null +++ b/gr-filter/python/qa_iir_filter.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python +# +# Copyright 2004,2007,2010 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 + +class test_iir_filter(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_iir_direct_001(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + fftaps = () + fbtaps = () + expected_result = (0, 0, 0, 0, 0, 0, 0, 0) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_002(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + fftaps = (2,) + fbtaps = (0,) + expected_result = (2, 4, 6, 8, 10, 12, 14, 16) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_003(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + fftaps = (2, 11) + fbtaps = (0, 0) + expected_result = (2, 15, 28, 41, 54, 67, 80, 93) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_004(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + fftaps = (2, 11) + fbtaps = (0, -1) + expected_result = (2, 13, 15, 26, 28, 39, 41, 52) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_005(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + fftaps = (2, 11, 0) + fbtaps = (0, -1, 3) + expected_result = (2, 13, 21, 59, 58, 186, 68, 583) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_006(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + expected_result = (2, 13, 21, 59, 58, 186, 68, 583) + fftaps = (2, 1) + fbtaps = (0, -1) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + fftaps = (2, 11, 0) + fbtaps = (0, -1, 3) + op.set_taps(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_007(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + expected_result = (2,2,5,5,8,8,11,11) + fftaps = (2, 1) + fbtaps = (0, -1) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + fftaps = (2,0,1) + fbtaps = (0, -1) + op.set_taps(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_iir_direct_008(self): + src_data = (1, 2, 3, 4, 5, 6, 7, 8) + expected_result = (2,4,4,10,18,14,26,56) + fftaps = (2,) + fbtaps = (0, 1) + src = gr.vector_source_f(src_data) + op = filter.iir_filter_ffd(fftaps, fbtaps) + fftaps_data = (1) + fbtaps = (0,0, -1,3) + op.set_taps(fftaps, fbtaps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual (expected_result, result_data) + +if __name__ == '__main__': + gr_unittest.run(test_iir_filter, "test_iir_filter.xml") + diff --git a/gr-filter/python/qa_interp_fir_filter.py b/gr-filter/python/qa_interp_fir_filter.py new file mode 100755 index 000000000..839330539 --- /dev/null +++ b/gr-filter/python/qa_interp_fir_filter.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python +# +# Copyright 2004,2007,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, gr_unittest +import filter_swig as filter +import math + +class test_interp_fir_filter(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_fff(self): + taps = [1, 10, 100, 1000, 10000] + src_data = (0, 2, 3, 5, 7, 11, 13, 17) + interpolation = 3 + xr = (0,0,0,0, + 2,20,200,2003,20030, + 300,3005,30050, + 500,5007,50070, + 700,7011,70110, + 1100,11013,110130, + 1300,13017,130170) + expected_result = tuple([float(x) for x in xr]) + + src = gr.vector_source_f(src_data) + op = filter.interp_fir_filter_fff(interpolation, taps) + dst = gr.vector_sink_f() + self.tb.connect(src, op) + self.tb.connect(op, dst) + self.tb.run() + result_data = dst.data() + L = min(len(result_data), len(expected_result)) + self.assertEqual(expected_result[0:L], result_data[0:L]) + +if __name__ == '__main__': + gr_unittest.run(test_interp_fir_filter, "test_interp_fir_filter.xml") + diff --git a/gr-filter/python/qa_pfb_arb_resampler.py b/gr-filter/python/qa_pfb_arb_resampler.py new file mode 100755 index 000000000..a4e22f4c0 --- /dev/null +++ b/gr-filter/python/qa_pfb_arb_resampler.py @@ -0,0 +1,95 @@ +#!/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_arb_resampler(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_fff_000(self): + N = 1000 # number of samples to use + fs = 1000 # baseband sampling rate + rrate = 1.123 # resampling rate + + nfilts = 32 + taps = filter.firdes.low_pass_2(nfilts, nfilts*fs, fs/2, fs/10, + attenuation_dB=80, + window=filter.firdes.WIN_BLACKMAN_hARRIS) + + freq = 100 + signal = gr.sig_source_f(fs, gr.GR_SIN_WAVE, freq, 1) + head = gr.head(gr.sizeof_float, N) + pfb = filter.pfb_arb_resampler_fff(rrate, taps) + snk = gr.vector_sink_f() + + self.tb.connect(signal, head, pfb, snk) + self.tb.run() + + Ntest = 50 + L = len(snk.data()) + t = map(lambda x: float(x)/(fs*rrate), xrange(L)) + + phase = 0.53013 + expected_data = map(lambda x: math.sin(2.*math.pi*freq*x+phase), t) + + dst_data = snk.data() + self.assertFloatTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 3) + + def test_ccf_000(self): + N = 1000 # number of samples to use + fs = 1000 # baseband sampling rate + rrate = 1.123 # resampling rate + + nfilts = 32 + taps = filter.firdes.low_pass_2(nfilts, nfilts*fs, fs/2, fs/10, + attenuation_dB=80, + window=filter.firdes.WIN_BLACKMAN_hARRIS) + + freq = 100 + signal = gr.sig_source_c(fs, gr.GR_SIN_WAVE, freq, 1) + head = gr.head(gr.sizeof_gr_complex, N) + pfb = filter.pfb_arb_resampler_ccf(rrate, taps) + snk = gr.vector_sink_c() + + self.tb.connect(signal, head, pfb, snk) + self.tb.run() + + Ntest = 50 + L = len(snk.data()) + t = map(lambda x: float(x)/(fs*rrate), xrange(L)) + + phase = 0.53013 + expected_data = map(lambda x: math.cos(2.*math.pi*freq*x+phase) + \ + 1j*math.sin(2.*math.pi*freq*x+phase), t) + + dst_data = snk.data() + self.assertComplexTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 3) + +if __name__ == '__main__': + gr_unittest.run(test_pfb_arb_resampler, "test_pfb_arb_resampler.xml") diff --git a/gr-filter/python/qa_pfb_channelizer.py b/gr-filter/python/qa_pfb_channelizer.py new file mode 100755 index 000000000..33d2b2188 --- /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 = filter.firdes.low_pass_2(1, ifs, 500, 50, + attenuation_dB=80, + window=filter.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") diff --git a/gr-filter/python/qa_pfb_decimator.py b/gr-filter/python/qa_pfb_decimator.py new file mode 100755 index 000000000..063845f63 --- /dev/null +++ b/gr-filter/python/qa_pfb_decimator.py @@ -0,0 +1,125 @@ +#!/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_decimator(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 + fs = 1000 # baseband sampling rate + ifs = M*fs # input samp rate to decimator + channel = 0 # Extract channel 0 + + taps = filter.firdes.low_pass_2(1, ifs, fs/2, fs/10, + attenuation_dB=80, + window=filter.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_decimator_ccf(M, taps, channel) + snk = gr.vector_sink_c() + + self.tb.connect(add, head, s2ss) + for i in xrange(M): + self.tb.connect((s2ss,i), (pfb,i)) + self.tb.connect(pfb, snk) + + self.tb.run() + + Ntest = 50 + L = len(snk.data()) + t = map(lambda x: float(x)/fs, xrange(L)) + + # Create known data as complex sinusoids for the baseband freq + # of the extracted channel is due to decimator output order. + phase = 0 + expected_data = map(lambda x: math.cos(2.*math.pi*freqs[2]*x+phase) + \ + 1j*math.sin(2.*math.pi*freqs[2]*x+phase), t) + + dst_data = snk.data() + + self.assertComplexTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 4) + + def test_001(self): + N = 1000 # number of samples to use + M = 5 # Number of channels + fs = 1000 # baseband sampling rate + ifs = M*fs # input samp rate to decimator + channel = 1 # Extract channel 0 + + taps = filter.firdes.low_pass_2(1, ifs, fs/2, fs/10, + attenuation_dB=80, + window=filter.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_decimator_ccf(M, taps, channel) + snk = gr.vector_sink_c() + + self.tb.connect(add, head, s2ss) + for i in xrange(M): + self.tb.connect((s2ss,i), (pfb,i)) + self.tb.connect(pfb, snk) + + self.tb.run() + + Ntest = 50 + L = len(snk.data()) + t = map(lambda x: float(x)/fs, xrange(L)) + + # Create known data as complex sinusoids for the baseband freq + # of the extracted channel is due to decimator output order. + phase = 6.15746 + expected_data = map(lambda x: math.cos(2.*math.pi*freqs[3]*x+phase) + \ + 1j*math.sin(2.*math.pi*freqs[3]*x+phase), t) + dst_data = snk.data() + + self.assertComplexTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 4) + +if __name__ == '__main__': + gr_unittest.run(test_pfb_decimator, "test_pfb_decimator.xml") diff --git a/gr-filter/python/qa_pfb_interpolator.py b/gr-filter/python/qa_pfb_interpolator.py new file mode 100755 index 000000000..5b84b7c64 --- /dev/null +++ b/gr-filter/python/qa_pfb_interpolator.py @@ -0,0 +1,71 @@ +#!/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_interpolator(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 + fs = 1000 # baseband sampling rate + ifs = M*fs # input samp rate to decimator + + taps = filter.firdes.low_pass_2(M, ifs, fs/2, fs/10, + attenuation_dB=80, + window=filter.firdes.WIN_BLACKMAN_hARRIS) + + freq = 100 + signal = gr.sig_source_c(fs, gr.GR_COS_WAVE, freq, 1) + head = gr.head(gr.sizeof_gr_complex, N) + pfb = filter.pfb_interpolator_ccf(M, taps) + snk = gr.vector_sink_c() + + self.tb.connect(signal, head, pfb) + self.tb.connect(pfb, snk) + + self.tb.run() + + Ntest = 50 + L = len(snk.data()) + t = map(lambda x: float(x)/ifs, xrange(L)) + + # Create known data as complex sinusoids at freq + # of the channel at the interpolated rate. + phase = 0.62833 + expected_data = map(lambda x: math.cos(2.*math.pi*freq*x+phase) + \ + 1j*math.sin(2.*math.pi*freq*x+phase), t) + + dst_data = snk.data() + + self.assertComplexTuplesAlmostEqual(expected_data[-Ntest:], dst_data[-Ntest:], 4) + +if __name__ == '__main__': + gr_unittest.run(test_pfb_interpolator, "test_pfb_interpolator.xml") diff --git a/gr-filter/python/qa_pfb_synthesizer.py b/gr-filter/python/qa_pfb_synthesizer.py new file mode 100755 index 000000000..e8164d268 --- /dev/null +++ b/gr-filter/python/qa_pfb_synthesizer.py @@ -0,0 +1,87 @@ +#!/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_synthesizer(gr_unittest.TestCase): + + def setUp(self): + self.tb = gr.top_block() + + def tearDown(self): + self.tb = None + + def test_000(self): + N = 10000 # number of samples to use + M = 5 # Number of channels + fs = 1000 # baseband sampling rate + ofs = M*fs # input samp rate to decimator + + taps = filter.firdes.low_pass_2(M, ofs, fs/2, fs/10, + attenuation_dB=80, + window=filter.firdes.WIN_BLACKMAN_hARRIS) + + signals = list() + freqs = [0, 100, 200, -200, -100] + for i in xrange(len(freqs)): + signals.append(gr.sig_source_c(fs, gr.GR_SIN_WAVE, freqs[i], 1)) + + head = gr.head(gr.sizeof_gr_complex, N) + pfb = filter.pfb_synthesizer_ccf(M, taps) + snk = gr.vector_sink_c() + + for i in xrange(M): + self.tb.connect(signals[i], (pfb,i)) + + self.tb.connect(pfb, head, snk) + + self.tb.run() + + Ntest = 1000 + L = len(snk.data()) + t = map(lambda x: float(x)/ofs, xrange(L)) + + # Create known data as sum of complex sinusoids at freqs + # of the output channels. + freqs = [-2200, -1100, 0, 1100, 2200] + expected_data = len(t)*[0,] + for i in xrange(len(t)): + expected_data[i] = math.cos(2.*math.pi*freqs[0]*t[i]) + \ + 1j*math.sin(2.*math.pi*freqs[0]*t[i]) + \ + math.cos(2.*math.pi*freqs[1]*t[i]) + \ + 1j*math.sin(2.*math.pi*freqs[1]*t[i]) + \ + math.cos(2.*math.pi*freqs[2]*t[i]) + \ + 1j*math.sin(2.*math.pi*freqs[2]*t[i]) + \ + math.cos(2.*math.pi*freqs[3]*t[i]) + \ + 1j*math.sin(2.*math.pi*freqs[3]*t[i]) + \ + math.cos(2.*math.pi*freqs[4]*t[i]) + \ + 1j*math.sin(2.*math.pi*freqs[4]*t[i]) + dst_data = snk.data() + + offset = 25 + self.assertComplexTuplesAlmostEqual(expected_data[2000-offset:2000-offset+Ntest], + dst_data[2000:2000+Ntest], 4) + +if __name__ == '__main__': + gr_unittest.run(test_pfb_synthesizer, "test_pfb_synthesizer.xml") diff --git a/gr-filter/python/qa_pm_remez.py b/gr-filter/python/qa_pm_remez.py new file mode 100755 index 000000000..a76e14c93 --- /dev/null +++ b/gr-filter/python/qa_pm_remez.py @@ -0,0 +1,188 @@ +#!/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 sys, math + +# ---------------------------------------------------------------- +# See optfir for an explanation of these. + +def stopband_atten_to_dev (atten_db): + """Convert a stopband attenuation in dB to an absolute value""" + return 10**(-atten_db/20) + +def passband_ripple_to_dev (ripple_db): + """Convert passband ripple spec expressed in dB to an absolute value""" + return (10**(ripple_db/20)-1)/(10**(ripple_db/20)+1) + +# ---------------------------------------------------------------- + +def remezord (fcuts, mags, devs, fsamp = 2): + ''' + FIR order estimator (lowpass, highpass, bandpass, mulitiband). + ''' + + # get local copies + fcuts = fcuts[:] + mags = mags[:] + devs = devs[:] + + for i in range (len (fcuts)): + fcuts[i] = float (fcuts[i]) / fsamp + + nf = len (fcuts) + nm = len (mags) + nd = len (devs) + nbands = nm + + if nm != nd: + raise ValueError, "Length of mags and devs must be equal" + + if nf != 2 * (nbands - 1): + raise ValueError, "Length of f must be 2 * len (mags) - 2" + + for i in range (len (mags)): + if mags[i] != 0: # if not stopband, get relative deviation + devs[i] = devs[i] / mags[i] + + # separate the passband and stopband edges + f1 = fcuts[0::2] + f2 = fcuts[1::2] + + n = 0 + min_delta = 2 + for i in range (len (f1)): + if f2[i] - f1[i] < min_delta: + n = i + min_delta = f2[i] - f1[i] + + if nbands == 2: + # lowpass or highpass case (use formula) + l = lporder (f1[n], f2[n], devs[0], devs[1]) + else: + # bandpass or multipass case + # try different lowpasses and take the worst one that + # goes through the BP specs + l = 0 + for i in range (1, nbands-1): + l1 = lporder (f1[i-1], f2[i-1], devs[i], devs[i-1]) + l2 = lporder (f1[i], f2[i], devs[i], devs[i+1]) + l = max (l, l1, l2) + + n = int (math.ceil (l)) - 1 # need order, not length for remez + + # cook up remez compatible result + ff = [0] + fcuts + [1] + for i in range (1, len (ff) - 1): + ff[i] *= 2 + + aa = [] + for a in mags: + aa = aa + [a, a] + + max_dev = max (devs) + wts = [1] * len(devs) + for i in range (len (wts)): + wts[i] = max_dev / devs[i] + + return (n, ff, aa, wts) + + +def lporder (freq1, freq2, delta_p, delta_s): + ''' + FIR lowpass filter length estimator. + ''' + df = abs (freq2 - freq1) + ddp = math.log10 (delta_p) + dds = math.log10 (delta_s) + + a1 = 5.309e-3 + a2 = 7.114e-2 + a3 = -4.761e-1 + a4 = -2.66e-3 + a5 = -5.941e-1 + a6 = -4.278e-1 + + b1 = 11.01217 + b2 = 0.5124401 + + t1 = a1 * ddp * ddp + t2 = a2 * ddp + t3 = a4 * ddp * ddp + t4 = a5 * ddp + + dinf=((t1 + t2 + a3) * dds) + (t3 + t4 + a6) + ff = b1 + b2 * (ddp - dds) + n = dinf / df - ff * df + 1 + return n + +# ---------------------------------------------------------------- + +class test_pm_remez(gr_unittest.TestCase): + + def setUp(self): + pass + + def tearDown(self): + pass + + def test_low_pass(self): + gain = 1 + Fs = 1 + freq1 = 0.1 + freq2 = 0.2 + passband_ripple_db = 0.01 + stopband_atten_db = 60 + + passband_dev = passband_ripple_to_dev(passband_ripple_db) + stopband_dev = stopband_atten_to_dev(stopband_atten_db) + desired_ampls = (gain, 0) + (n, fo, ao, w) = remezord([freq1, freq2], desired_ampls, + [passband_dev, stopband_dev], Fs) + new_taps = filter.pm_remez(n + 2, fo, ao, w, "bandpass") + + known_taps = (-0.0008370135734511828, -0.0006622211673134374, + 0.0008501079576365787, 0.003059609130249229, + 0.003202235537205373, -0.001000899296974219, + -0.007589728680590891, -0.009790921118281865, + -0.001524210202628562, 0.014373535837200111, + 0.02392881326993834, 0.011798133085019008, + -0.021954446348997188, -0.05293436740264934, + -0.04375787096766848, 0.028038890498420392, + 0.14612655590172896, 0.25738578419108626, + 0.302967004188747, 0.25738578419108626, + 0.14612655590172896, 0.028038890498420392, + -0.04375787096766848, -0.05293436740264934, + -0.021954446348997188, 0.011798133085019008, + 0.02392881326993834, 0.014373535837200111, + -0.001524210202628562, -0.009790921118281865, + -0.007589728680590891, -0.001000899296974219, + 0.003202235537205373, 0.003059609130249229, + 0.0008501079576365787, -0.0006622211673134374, + -0.0008370135734511828) + + self.assertFloatTuplesAlmostEqual(known_taps, new_taps, 5) + +if __name__ == '__main__': + gr_unittest.run(test_pm_remez, "test_pm_remez.xml") + diff --git a/gr-filter/python/qa_rational_resampler.py b/gr-filter/python/qa_rational_resampler.py new file mode 100755 index 000000000..eb86ef542 --- /dev/null +++ b/gr-filter/python/qa_rational_resampler.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python +# +# Copyright 2005,2006,2007,2010 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 +import random +import sys + + +def random_floats(n): + r = [] + for x in xrange(n): + # r.append(float(random.randint(-32768, 32768))) + r.append(float(random.random())) + return tuple(r) + + +def reference_dec_filter(src_data, decim, taps): + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.fir_filter_fff(decim, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + tb = None + return result_data + +def reference_interp_filter(src_data, interp, taps): + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.interp_fir_filter_fff(interp, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + result_data = dst.data() + tb = None + return result_data + +def reference_interp_dec_filter(src_data, interp, decim, taps): + tb = gr.top_block() + src = gr.vector_source_f(src_data) + up = filter.interp_fir_filter_fff(interp, (1,)) + dn = filter.fir_filter_fff(decim, taps) + dst = gr.vector_sink_f() + tb.connect(src, up, dn, dst) + tb.run() + result_data = dst.data() + tb = None + return result_data + + +class test_rational_resampler (gr_unittest.TestCase): + + def setUp(self): + random.seed(0) + + def tearDown(self): + pass + + + def test_000_1_to_1(self): + taps = (-4, 5) + src_data = (234, -4, 23, -56, 45, 98, -23, -7) + xr = (1186, -112, 339, -460, -167, 582) + expected_result = tuple([float(x) for x in xr]) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(1, 1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + result_data = dst.data() + self.assertEqual(expected_result, result_data) + + def test_001_interp(self): + taps = [1, 10, 100, 1000, 10000] + src_data = (0, 2, 3, 5, 7, 11, 13, 17) + interpolation = 3 + xr = (0,2,20,200,2003,20030, + 300,3005,30050, + 500,5007,50070, + 700,7011,70110, + 1100,11013,110130, + 1300,13017,130170, + 1700.0,17000.0,170000.0) + expected_result = tuple([float(x) for x in xr]) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(interpolation, 1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + result_data = dst.data() + self.assertEqual(expected_result, result_data) + + def test_002_interp(self): + taps = random_floats(31) + src_data = random_floats(10000) + interpolation = 3 + + expected_result = reference_interp_filter(src_data, interpolation, taps) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(interpolation, 1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + result_data = dst.data() + + N = 1000 + offset = len(taps)-1 + self.assertEqual(expected_result[offset:offset+N], result_data[0:N]) + + def xtest_003_interp(self): + taps = random_floats(9) + src_data = random_floats(10000) + decimation = 3 + + expected_result = reference_dec_filter(src_data, decimation, taps) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(1, decimation, taps) + dst = gr.vector_sink_f() + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + result_data = dst.data() + + N = 10 + offset = 10#len(taps)-1 + print expected_result[100+offset:100+offset+N] + print result_data[100:100+N] + #self.assertEqual(expected_result[offset:offset+N], result_data[0:N]) + + # FIXME disabled. Triggers hang on SuSE 10.0 + def xtest_004_decim_random_vals(self): + MAX_TAPS = 9 + MAX_DECIM = 7 + OUTPUT_LEN = 9 + + random.seed(0) # we want reproducibility + + for ntaps in xrange(1, MAX_TAPS + 1): + for decim in xrange(1, MAX_DECIM+1): + for ilen in xrange(ntaps + decim, ntaps + OUTPUT_LEN*decim): + src_data = random_floats(ilen) + taps = random_floats(ntaps) + expected_result = reference_dec_filter(src_data, decim, taps) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(1, decim, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + tb = None + result_data = dst.data() + + L1 = len(result_data) + L2 = len(expected_result) + L = min(L1, L2) + if False: + sys.stderr.write('delta = %2d: ntaps = %d decim = %d ilen = %d\n' % (L2 - L1, ntaps, decim, ilen)) + sys.stderr.write(' len(result_data) = %d len(expected_result) = %d\n' % + (len(result_data), len(expected_result))) + self.assertEqual(expected_result[0:L], result_data[0:L]) + + + # FIXME disabled. Triggers hang on SuSE 10.0 + def xtest_005_interp_random_vals(self): + MAX_TAPS = 9 + MAX_INTERP = 7 + INPUT_LEN = 9 + + random.seed(0) # we want reproducibility + + for ntaps in xrange(1, MAX_TAPS + 1): + for interp in xrange(1, MAX_INTERP+1): + for ilen in xrange(ntaps, ntaps + INPUT_LEN): + src_data = random_floats(ilen) + taps = random_floats(ntaps) + expected_result = reference_interp_filter(src_data, interp, taps) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(interp, 1, taps) + dst = gr.vector_sink_f() + tb.connect(src, op, dst) + tb.run() + tb = None + result_data = dst.data() + L1 = len(result_data) + L2 = len(expected_result) + L = min(L1, L2) + #if True or abs(L1-L2) > 1: + if False: + sys.stderr.write('delta = %2d: ntaps = %d interp = %d ilen = %d\n' % (L2 - L1, ntaps, interp, ilen)) + #sys.stderr.write(' len(result_data) = %d len(expected_result) = %d\n' % + # (len(result_data), len(expected_result))) + #self.assertEqual(expected_result[0:L], result_data[0:L]) + # FIXME check first ntaps+1 answers + self.assertEqual(expected_result[ntaps+1:L], result_data[ntaps+1:L]) + + def test_006_interp_decim(self): + taps = random_floats(31) + src_data = random_floats(10000) + interp = 3 + decimation = 2 + + expected_result = reference_interp_dec_filter(src_data, interp, decimation, taps) + + tb = gr.top_block() + src = gr.vector_source_f(src_data) + op = filter.rational_resampler_base_fff(interp, decimation, taps) + dst = gr.vector_sink_f() + tb.connect(src, op) + tb.connect(op, dst) + tb.run() + result_data = dst.data() + + N = 1000 + offset = len(taps)/2 + self.assertFloatTuplesAlmostEqual(expected_result[offset:offset+N], result_data[0:N], 5) + + +if __name__ == '__main__': + # FIXME: Disabled, see ticket:210 + gr_unittest.run(test_rational_resampler, "test_rational_resampler.xml") + diff --git a/gr-filter/python/qa_single_pole_iir.py b/gr-filter/python/qa_single_pole_iir.py new file mode 100755 index 000000000..3608c77f9 --- /dev/null +++ b/gr-filter/python/qa_single_pole_iir.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python +# +# Copyright 2005,2007,2010 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 + +class test_single_pole_iir_filter(gr_unittest.TestCase): + + def setUp (self): + self.tb = gr.top_block () + + def tearDown (self): + self.tb = None + + def test_ff_001(self): + src_data = (0, 1000, 2000, 3000, 4000, 5000) + expected_result = src_data + src = gr.vector_source_f(src_data) + op = filter.single_pole_iir_filter_ff(1.0) + dst = gr.vector_sink_f() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data) + + def test_ff_002(self): + src_data = (0, 1000, 2000, 3000, 4000, 5000) + expected_result = (0, 125, 359.375, 689.453125, 1103.271484, 1590.36255) + src = gr.vector_source_f(src_data) + op = filter.single_pole_iir_filter_ff(0.125) + dst = gr.vector_sink_f() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data, 3) + + def test_ff_003(self): + block_size = 2 + src_data = (0, 1000, 2000, 3000, 4000, 5000) + expected_result = (0, 125, 250, 484.375, 718.75, 1048.828125) + src = gr.vector_source_f(src_data) + s2p = gr.serial_to_parallel(gr.sizeof_float, block_size) + op = filter.single_pole_iir_filter_ff (0.125, block_size) + p2s = gr.parallel_to_serial(gr.sizeof_float, block_size) + dst = gr.vector_sink_f() + self.tb.connect(src, s2p, op, p2s, dst) + self.tb.run() + result_data = dst.data() + self.assertFloatTuplesAlmostEqual(expected_result, result_data, 3) + + def test_cc_001(self): + src_data = (0+0j, 1000+1000j, 2000+2000j, 3000+3000j, 4000+4000j, 5000+5000j) + expected_result = src_data + src = gr.vector_source_c(src_data) + op = filter.single_pole_iir_filter_cc(1.0) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_result, result_data) + + def test_cc_002(self): + src_data = (complex(0,0), complex(1000,-1000), complex(2000,-2000), + complex(3000,-3000), complex(4000,-4000), complex(5000,-5000)) + expected_result = (complex(0,0), complex(125,-125), complex(359.375,-359.375), + complex(689.453125,-689.453125), complex(1103.271484,-1103.271484), + complex(1590.36255,-1590.36255)) + src = gr.vector_source_c(src_data) + op = filter.single_pole_iir_filter_cc(0.125) + dst = gr.vector_sink_c() + self.tb.connect(src, op, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_result, result_data, 3) + + def test_cc_003(self): + block_size = 2 + src_data = (complex(0,0), complex(1000,-1000), complex(2000,-2000), + complex(3000,-3000), complex(4000,-4000), complex(5000,-5000)) + expected_result = (complex(0,0), complex(125,-125), complex(250,-250), + complex(484.375,-484.375), complex(718.75,-718.75), + complex(1048.828125,-1048.828125)) + src = gr.vector_source_c(src_data) + s2p = gr.serial_to_parallel(gr.sizeof_gr_complex, block_size) + op = filter.single_pole_iir_filter_cc(0.125, block_size) + p2s = gr.parallel_to_serial(gr.sizeof_gr_complex, block_size) + dst = gr.vector_sink_c() + self.tb.connect(src, s2p, op, p2s, dst) + self.tb.run() + result_data = dst.data() + self.assertComplexTuplesAlmostEqual(expected_result, result_data, 3) + +if __name__ == '__main__': + gr_unittest.run(test_single_pole_iir_filter, "test_single_pole_iir_filter.xml") + diff --git a/gr-filter/python/rational_resampler.py b/gr-filter/python/rational_resampler.py new file mode 100644 index 000000000..312b011d3 --- /dev/null +++ b/gr-filter/python/rational_resampler.py @@ -0,0 +1,129 @@ +# +# Copyright 2005,2007 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, gru + +_plot = None + +def design_filter(interpolation, decimation, fractional_bw): + """ + Given the interpolation rate, decimation rate and a fractional bandwidth, + design a set of taps. + + @param interpolation: interpolation factor + @type interpolation: integer > 0 + @param decimation: decimation factor + @type decimation: integer > 0 + @param fractional_bw: fractional bandwidth in (0, 0.5) 0.4 works well. + @type fractional_bw: float + @returns: sequence of numbers + """ + + if fractional_bw >= 0.5 or fractional_bw <= 0: + raise ValueError, "Invalid fractional_bandwidth, must be in (0, 0.5)" + + beta = 5.0 + trans_width = 0.5 - fractional_bw + mid_transition_band = 0.5 - trans_width/2 + + taps = filter.firdes.low_pass(interpolation, # gain + 1, # Fs + mid_transition_band/interpolation, # trans mid point + trans_width/interpolation, # transition width + filter.firdes.WIN_KAISER, + beta) # beta + return taps + + + +class _rational_resampler_base(gr.hier_block2): + """ + base class for all rational resampler variants. + """ + def __init__(self, resampler_base, + interpolation, decimation, taps=None, fractional_bw=None): + """ + Rational resampling polyphase FIR filter. + + Either taps or fractional_bw may be specified, but not both. + If neither is specified, a reasonable default, 0.4, is used as + the fractional_bw. + + @param interpolation: interpolation factor + @type interpolation: integer > 0 + @param decimation: decimation factor + @type decimation: integer > 0 + @param taps: optional filter coefficients + @type taps: sequence + @param fractional_bw: fractional bandwidth in (0, 0.5), measured at final freq (use 0.4) + @type fractional_bw: float + """ + + if not isinstance(interpolation, int) or interpolation < 1: + raise ValueError, "interpolation must be an integer >= 1" + + if not isinstance(decimation, int) or decimation < 1: + raise ValueError, "decimation must be an integer >= 1" + + if taps is None and fractional_bw is None: + fractional_bw = 0.4 + + d = gru.gcd(interpolation, decimation) + interpolation = interpolation // d + decimation = decimation // d + + if taps is None: + taps = design_filter(interpolation, decimation, fractional_bw) + + resampler = resampler_base(interpolation, decimation, taps) + gr.hier_block2.__init__(self, "rational_resampler", + gr.io_signature(1, 1, resampler.input_signature().sizeof_stream_item(0)), + gr.io_signature(1, 1, resampler.output_signature().sizeof_stream_item(0))) + + self.connect(self, resampler, self) + + +class rational_resampler_fff(_rational_resampler_base): + def __init__(self, interpolation, decimation, taps=None, fractional_bw=None): + """ + Rational resampling polyphase FIR filter with + float input, float output and float taps. + """ + _rational_resampler_base.__init__(self, filter.rational_resampler_base_fff, + interpolation, decimation, taps, fractional_bw) + +class rational_resampler_ccf(_rational_resampler_base): + def __init__(self, interpolation, decimation, taps=None, fractional_bw=None): + """ + Rational resampling polyphase FIR filter with + complex input, complex output and float taps. + """ + _rational_resampler_base.__init__(self, filter.rational_resampler_base_ccf, + interpolation, decimation, taps, fractional_bw) + +class rational_resampler_ccc(_rational_resampler_base): + def __init__(self, interpolation, decimation, taps=None, fractional_bw=None): + """ + Rational resampling polyphase FIR filter with + complex input, complex output and complex taps. + """ + _rational_resampler_base.__init__(self, filter.rational_resampler_base_ccc, + interpolation, decimation, taps, fractional_bw) |