# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

import numpy as np

from scipy.stats import norm, laplace, rayleigh

rndm = gr.random() # instance of gnuradio random class (gr::random)

#*** GENERATE DATA ***#

laplace_expected = np.zeros(laplace_num_bins-1)

for k in range(len(uniform_hist[0])):

for k in range(len(gauss_hist[0])):

gauss_expected[k] = float(norm.cdf(gauss_hist[1][k+1])-norm.cdf(gauss_hist[1][k]))*num_tests

#*** PLOT HISTOGRAMS AND EXPECTATIONS TAKEN FROM SCIPY ***#

plt.figure(1)

plt.legend(['histogram gr::random','calculation scipy'],loc=1)

plt.subplot(2,1,2)

plt.xlabel('Bins'), plt.ylabel('Relative deviation'), plt.title('Uniform: Relative deviation to scipy')

plt.figure(2)

plt.legend(['histogram gr::random','calculation scipy'],loc=1)

plt.subplot(2,1,2)

plt.xlabel('Bins'), plt.ylabel('Relative deviation'), plt.title('Gauss: Relative deviation to scipy')

plt.figure(3)

plt.subplot(2,1,2)

plt.xlabel('Bins'), plt.ylabel('Relative deviation'), plt.title('Rayleigh: Relative deviation to scipy')

plt.figure(4)

plt.legend(['histogram gr::random','calculation scipy'],loc=1)

plt.subplot(2,1,2)

plt.xlabel('Bins'), plt.ylabel('Relative deviation'), plt.title('Laplace: Relative deviation to scipy')

plt.show()

# Title: Affinity Set Test

##################################################

from gnuradio import eng_notation

from gnuradio import gr

from gnuradio import blocks

tb.start()

while(1):

if(len(ret) == 0):

tb.stop()

sys.exit(0)

try:

n = int(ret)

except ValueError:

else:

tb.filter_filt_0.set_processor_affinity([n,])

Reads output from run_synthetic.py and runs gnuplot showing

GFLOPS as f(npipes, nstages)

"""

import re

import sys

"""

Run synthetic.py for npipes in [1,16], nstages in [1,16]

"""

import re

import sys

# We'd like each run of synthetic to take ~10 seconds

desired_time_per_run = 10

est_gflops_avail = min(nstages * npipes, ncores) * gflops

nsamples = int(nsamples * 1e9)

cmd = "./synthetic.py -m -s %d -p %d -N %d\n" % (nstages, npipes, nsamples)

# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

#

from gnuradio import gr, eng_notation

from gnuradio import blocks, filter

from gnuradio.eng_arg import eng_float, intx

gr.hier_block2.__init__(self, "pipeline",

gr.io_signature(1, 1, gr.sizeof_float),

gr.io_signature(0, 0, 0))

upstream = self

for i in range(nstages):

op = filter.fir_filter_fff(1, taps)

start = os.times()

tb.run()

stop = os.times()

user, sys, childrens_user, childrens_sys, real = delta

total_user = user + childrens_user

total_sys = sys + childrens_sys

if tb.machine_readable:

else:

if __name__ == "__main__":

tb = top()

time_it(tb)

except KeyboardInterrupt:

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, gru, eng_notation, filter

from gnuradio import audio

from gnuradio import analog

if args.volume is None:

g = self.volume_range()

# set initial values

def set_vol (self, vol):

g = self.volume_range()

self.vol = max(g[0], min(g[1], vol))

def volume_range(self):

return (-20.0, 0.0, 0.5)

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from gnuradio import blocks

from argparse import ArgumentParser

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from gnuradio import blocks

from argparse import ArgumentParser

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, audio

from gnuradio import blocks

from argparse import ArgumentParser

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from argparse import ArgumentParser

import sys

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from gnuradio import blocks

from gnuradio.eng_arg import eng_float, intx

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from gnuradio import blocks

from argparse import ArgumentParser

#!/usr/bin/env python

from gnuradio import gr

from gnuradio import blocks

import argparse

res = format_results(func.__name__, t)

return res

except AttributeError:

return None

def main():

default=20,

help='Number of iterations [default: %(default)s]')

parser.add_argument('--tests', type=int, nargs='*',

help='A list of tests to run; can be a single test or a \

space-separated list.')

parser.add_argument('--list', action='store_true',

args = parser.parse_args()

if(args.list):

sys.exit(0)

N = int(args.nitems)

new_table(conn, label)

if args.all:

else:

tests = args.tests

#!/usr/bin/env python

import sys, math

import argparse

from volk_test_funcs import *

# Plot the results

i = 0

for t in (table_data):

ydata = []

if(args.percent != t):

# makes x values for this data set placement

# width of bars depends on number of comparisons

x1 = [x + i*wdth for x in x0]

i += 1

else:

# makes x values for this data set placement

# width of bars depends on number of comparisons

x1 = [x + i*wdth for x in x0]

i += 1

#!/usr/bin/env python

from gnuradio import gr

from gnuradio import blocks

import math, sys, os, time

self.snks = []

self.head = blocks.head(isizeof, N)

self.srcs.append(blocks.null_source(isizeof))

self.snks.append(blocks.null_sink(osizeof))

self.connect(self.srcs[0], self.head, (self.op,0))

self.connect(self.srcs[n], (self.op,n))

self.connect((self.op,n), self.snks[n])

def timeit(tb, iterations):

'''

r = gr.enable_realtime_scheduling()

if r != gr.RT_OK:

times = []

start_time = time.time()

tb.run()

end_time = time.time()

#!/usr/bin/env python

from gnuradio import gr

from gnuradio import blocks

import argparse

res = format_results(func.__name__, t)

return res

except AttributeError:

return None

def main():

default=20,

help='Number of iterations [default: %(default)s]')

parser.add_argument('--tests', type=int, nargs='*',

help='A list of tests to run; can be a single test or a \

space-separated list.')

parser.add_argument('--list', action='store_true',

args = parser.parse_args()

if(args.list):

sys.exit(0)

N = int(args.nitems)

new_table(conn, label)

if args.all:

else:

tests = args.tests

static void create_and_stash_initial_sptr(gr::hier_block2 *p);

static void cancel_initial_sptr(gr::hier_block2 *p);

};

/*

* \brief New! Improved! Standard method to get/create the

# Handle the generated constants

########################################################################

execute_process(COMMAND ${PYTHON_EXECUTABLE} -c

OUTPUT_VARIABLE BUILD_DATE OUTPUT_STRIP_TRAILING_WHITESPACE

)

message(STATUS "Loading build date ${BUILD_DATE} into constants...")

########################################################################

include_directories(${GNURADIO_RUNTIME_INCLUDE_DIRS}

${CMAKE_CURRENT_SOURCE_DIR}

${Boost_INCLUDE_DIRS}

)

# Boston, MA 02110-1301, USA.

#

import math

import sys

for i in range (nentries):

a = (i * incx) + min_x

b = ((i + 1) * incx) + min_x

abs_error = c+m*a-f(a)

r.append ((m, c, abs_error))

return r

${CMAKE_CURRENT_SOURCE_DIR}/unv_template.h.t

${CMAKE_CURRENT_SOURCE_DIR}/unv_template.cc.t

${CMAKE_CURRENT_SOURCE_DIR}/unv_qa_template.cc.t

"import os, sys; srcdir='${CMAKE_CURRENT_SOURCE_DIR}'; sys.path.append(srcdir); os.environ['srcdir']=srcdir; from generate_unv import main; main()"

WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}

VERBATIM

#endif /* INCLUDED_PMT_SERIAL_TAGS_H */

"""

import sys, os, re

if __name__ == '__main__':

if len(sys.argv) != 3:

exit()

input_scm_file, output_hdr_file = sys.argv[1:]

enums = list()

"""

Generate code for uniform numeric vectors

"""

import re, os, os.path

# set srcdir to the directory that contains Makefile.am

try:

srcdir = os.environ['srcdir']

srcdir = "."

srcdir = srcdir + '/'

s_map.erase(pos);

return sptr;

}

"""Misc utilities used at build time

"""

import re, os, os.path

from build_utils_codes import *

# set srcdir to the directory that contains Makefile.am

try:

srcdir = os.environ['srcdir']

srcdir = "."

srcdir = srcdir + '/'

do_makefile = False

else:

do_makefile = True

do_makefile = False

# set do_sources to either true or false dependeing on the environment

do_sources = False

else:

do_sources = True

do_sources = True

name_dict = {}

# we return everything between the penultimate . and .t

mo = re.search (r'\.([a-z]+)\.t$', template_name)

if not mo:

return mo.group (1)

def open_src (name, mode):

#

# Copyright 2004 Free Software Foundation, Inc.

#

GNU Radio is licensed under the GNU General Public License (GPL) version 3. All of the code is copyright of the Free Software Foundation.

"""

# This file makes gnuradio a package

# The docstring will be associated with the top level of the package.

is currently the only supported transport.

"""

import exceptions

"""

"""

Constructor for creating a ControlPort connection to a specified host / port

self.client = None

from gnuradio.ctrlport.RPCConnection import RPCMethods

from gnuradio.ctrlport.RPCConnectionThrift import RPCConnectionThrift

if rpcmethod == 'thrift':

#print("making RPCConnectionThrift")

if not blockingcallback is None:

blockingcallback()

else:

raise exceptions.ValueError()

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

from gnuradio import blocks

from gnuradio import filter

from PyQt4 import QtGui, QtCore

import sip

except ImportError:

sys.exit(1)

class GrDataPlotParent(gr.top_block, QtGui.QWidget):

self.layout.removeWidget(self.py_window)

self.disconnect(self.thr, (self.snk, 0))

self.disconnect(self.src[0], self.thr)

self.disconnect(self.src[n], (self.snk,n))

self._ncons = nconnections

self._last_data = []

self.src = []

self.set_line_label(n, self.knobnames[n])

self._last_data.append(int(self._npts)*[0,])

if(self._npts != npts):

# Adjust buffers to accommodate new settings

if(npts < self._npts):

if(self._data_len[n] < npts):

self._last_data[n] = self._last_data[n][0:npts]

if(self._stripchart):

# Update the plot data depending on type

if(type(data[n]) == list):

data[n] = self.data_to_complex(data[n])

if(len(data[n]) > self._npts):

self._last_data[n].append(data[n])

self.src[n].set_data(self._last_data[n])

else:

if(type(data[n]) != list):

data[n] = [data[n],]

data[n] = self.data_to_complex(data[n])

self.snk.set_line_label(n, self.knobnames[n])

def __init__(self, uid, parent, x, y, xsize, ysize,

headers=['Statistic Key ( Source Block :: Stat Name ) ',

'Curent Value', 'Units', 'Description']):

self.uid = uid

self.treeWidget = QtGui.QTreeWidget(parent)

self.treeWidget.setColumnCount(len(headers))

# itemKey is the text in the first column of a QTreeWidgetItem

itemKey = str(item.text(0))

# This key was found in the tree, update its values.

foundKeys.append(itemKey)

deleteKeys.append(itemKey)

# Add items to tree that are not currently in the tree.

if k not in foundKeys:

v = knobs[k].value

if(type(v) == ControlPort.complex):

#

# Copyright 2015 Free Software Foundation, Inc.

#

# Boston, MA 02110-1301, USA.

#

from thrift import Thrift

from thrift.transport import TSocket

from thrift.transport import TTransport

import pmt

import sys

def __init__(self, host, port):

self.tsocket = TSocket.TSocket(host, port)

self.transport = TTransport.TBufferedTransport(self.tsocket)

"""

class RPCConnectionThrift(RPCConnection.RPCConnection):

def __init__(self, key, value=None, ktype=0):

(self.key, self.value, self.ktype) = (key, value, ktype)

def properties(self, *args):

knobprops = self.thriftclient.radio.properties(*args)

#print("key:", key, "value:", knobprop, "type:", knobprop.type)

knobprops[key].min = self.unpackKnob(key, knobprop.min)

knobprops[key].max = self.unpackKnob(key, knobprop.max)

def getKnobs(self, *args):

result = {}

#print("key:", key, "value:", knob, "type:", knob.type)

result[key] = self.unpackKnob(key, knob)

# If complex, convert to Python complex

# FIXME: better list iterator way to handle this?

if(knob.type == self.BaseTypes.C32VECTOR):

result[key].value[i] = complex(result[key].value[i].re,

result[key].value[i].im)

return result

def getKnobsRaw(self, *args):

result = {}

#print("key:", key, "value:", knob, "type:", knob.type)

result[key] = knob

return result

def getRe(self,*args):

result = {}

result[key] = self.unpackKnob(key, knob)

return result

if(type(*args) == dict):

a = dict(*args)

result = {}

result[key] = self.packKnob(knob)

self.thriftclient.radio.setKnobs(result)

elif(type(*args) == list or type(*args) == tuple):

#

# Copyright 2012 Free Software Foundation, Inc.

#

# import swig generated symbols into the ctrlport namespace

#from ctrlport_swig import *

# import any pure python here

#import GNURadio

# Boston, MA 02110-1301, USA.

#

import sys, time, re, pprint

import random,math,operator

try:

try:

from matplotlib.backends.backend_qt4agg import NavigationToolbar2QT as NavigationToolbar

except ImportError:

sys.exit(1)

from matplotlib.figure import Figure

except ImportError:

sys.exit(1)

from PyQt4 import QtCore,Qt

class DataTable(QtGui.QWidget):

def update(self):

def closeEvent(self, event):

self.timer = None

self.clockSel.setCurrentIndex(self.clockSelIdx)

self.prevent_clock_change = False

except:

nodes_stream = self.G_stream.nodes()

nodes_msg = self.G_msg.nodes()

self.parent.statusBar().showMessage("Current GNU Radio Control Port Query Latency: %f ms"%\

(latency*1000))

sys.stderr.write("gr-perf-monitorx: radio.getKnobs threw exception ({0}).\n".format(e))

if(type(self.parent) is MAINWindow):

# Find window of connection

return;

idx = self.clockSel.currentIndex();

clk = self.clocks.values()[idx]

k = self.radioclient.getKnobs([self.clockKey]);

k[self.clockKey].value = clk;

km = {};

# Boston, MA 02110-1301, USA.

#

import sys, subprocess, re, signal, time, atexit, os

from gnuradio import gr

def __init__(self,tool="gr-ctrlport-monitor"):

self.started = False

self.tool = tool

atexit.register(self.shutdown)

gr.prefs().singleton().set_bool("PerfCounters","on",True);

gr.prefs().singleton().set_bool("PerfCounters","export",True);

except:

def __del__(self):

if(self.started):

self.stop()

def start(self):

try:

cmd = map(lambda a: [self.tool,

re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)",a).group(1),

re.search("-p (\d+)",a).group(1)],

gr.rpcmanager_get().endpoints())[0]

self.proc = subprocess.Popen(cmd);

self.started = True

except:

self.proc = None

def stop(self):

if(self.proc):

if(self.proc.returncode == None):

self.proc.terminate()

else:

def shutdown(self):

if(self.started):

self.stop()

'''

Add support for engineering notation to argparse.ArgumentParser

'''

import argparse

from gnuradio import eng_notation

"""

Display numbers as strings using engineering notation.

"""

scale_factor = {}

scale_factor['E'] = 1e18

try:

scale = 1.0

suffix = value[-1]

return float (value[0:-1]) * scale_factor[suffix]

return float (value)

except:

#

'''Add support for engineering notation to optparse.OptionParser'''

from copy import copy

from optparse import Option, OptionValueError

def check_eng_float (option, opt, value):

try:

GR_PYTHON_INSTALL(FILES

__init__.py

tag_utils.py

packet_utils.py

gateway.py

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)

endforeach(py_qa_test_file)

endif(ENABLE_TESTING)

"""

Core contents.

"""

# This is the main GNU Radio python module.

# We pull the swig output and the other modules into the gnuradio.gr namespace

# If gnuradio is installed then the swig output will be in this directory.

# Otherwise it will reside in ../../../swig.

try:

except ImportError:

dirname, filename = os.path.split(os.path.abspath(__file__))

__path__.append(os.path.join(dirname, "..", "..", "..", "swig"))

# Force the preference database to be initialized

prefs = prefs.singleton

#

# Copyright 2004 Free Software Foundation, Inc.

#

# Boston, MA 02110-1301, USA.

#

import numpy

########################################################################

# Magic to turn pointers into numpy arrays

# http://docs.scipy.org/doc/numpy/reference/arrays.interface.html

########################################################################

def pointer_to_ndarray(addr, dtype, nitems):

__array_interface__ = {

'data' : (int(addr), False),

'typestr' : dtype.base.str,

#ensure that the sigs are iterable dtypes

def sig_to_dtype_sig(sig):

if sig is None: sig = ()

self.__in_sig = sig_to_dtype_sig(in_sig)

self.__out_sig = sig_to_dtype_sig(out_sig)

#cache the ranges to iterate when dispatching work

#convert the signatures into gr.io_signatures

def sig_to_gr_io_sigv(sig):

#

# Copyright 2005 Free Software Foundation, Inc.

#

if _version_info[0:2] == (2, 3):

#print "Importing gr_threading_23"

elif _version_info[0:2] == (2, 4):

#print "Importing gr_threading_24"

else:

# assume the patch was applied...

#print "Importing system provided threading"

# It's been patched to fix a problem with join, where a KeyboardInterrupt

# caused a lock to be left in the acquired state.

import sys as _sys

try:

except ImportError:

del _sys.modules[__name__]

raise

from time import time as _time, sleep as _sleep

from traceback import print_exc as _print_exc

'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',

'Timer', 'setprofile', 'settrace']

# Debug support (adapted from ihooks.py).

# Internal methods used by condition variables

self.__block.acquire()

self.__count = count

self.__owner = owner

if __debug__:

def release(self):

if self._Semaphore__value >= self._initial_value:

return _Semaphore.release(self)

def enumerate():

_active_limbo_lock.acquire()

_active_limbo_lock.release()

return active

def run(self):

while self.count > 0:

item = self.queue.get()

self.count = self.count - 1

NP = 3

# It's been patched to fix a problem with join, where a KeyboardInterrupt

# caused a lock to be left in the acquired state.

import sys as _sys

try:

except ImportError:

del _sys.modules[__name__]

raise

'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',

'Timer', 'setprofile', 'settrace', 'local']

# Debug support (adapted from ihooks.py).

# Internal methods used by condition variables

self.__block.acquire()

self.__count = count

self.__owner = owner

if __debug__:

def release(self):

if self._Semaphore__value >= self._initial_value:

return _Semaphore.release(self)

def enumerate():

_active_limbo_lock.acquire()

_active_limbo_lock.release()

return active

# module, or from the python fallback

try:

except ImportError:

from _threading_local import local

def run(self):

while self.count > 0:

item = self.queue.get()

self.count = self.count - 1

NP = 3

#

# Copyright 2006,2007,2014 Free Software Foundation, Inc.

#

import functools

import pmt

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr

import pmt

assert(len(offsets) == len(lengths))

for offset, length in zip(offsets, lengths):

tag = gr.tag_t()

tag.key = pmt.string_to_symbol(tagname)

tags.append(tag)

return tags

if pos in lengths:

if in_packet:

print("Got tag at pos {0} current packet_pos is {1}".format(pos, packet_pos))

packet_pos = -1

packet_length = lengths[pos]

in_packet = True

for packet in packets:

data.extend(packet)

tag = gr.tag_t()

tag.key = pmt.string_to_symbol(tsb_tag_key)

tags.append(tag)

offset = offset + len(packet)

return data, tags

This is a proof of concept implementation, will likely change significantly.

"""

class pubsub(dict):

def __init__(self):

def __missing__(self, key, value=None):

def __setitem__(self, key, val):

def __getitem__(self, key):

def publish(self, key, publisher):

if self._proxies[key] is not None:

(p, newkey) = self._proxies[key]

p.publish(newkey, publisher)

self._publishers[key] = publisher

def subscribe(self, key, subscriber):

if self._proxies[key] is not None:

(p, newkey) = self._proxies[key]

p.subscribe(newkey, subscriber)

self._subscribers[key].remove(subscriber)

def proxy(self, key, p, newkey=None):

def unproxy(self, key):

self._proxies[key] = None

o = pubsub()

# Non-existent key gets auto-created with None value

# Add some subscribers

# First is a bare function

def print_len(x):

o.subscribe('foo', print_len)

# The second is a class member function

class subber(object):

s = subber('param')

o.subscribe('foo', s.printer)

# The third is a lambda function

# Update key 'foo', will notify subscribers

o['foo'] = 'bar';

# Remove first subscriber

o.unsubscribe('foo', print_len)

# Update now will only trigger second and third subscriber

o['foo'] = 'bar2';

# Publish a key as a function, in this case, a lambda function

o.publish('baz', lambda : 42)

# Unpublish the key

o.unpublish('baz')

# This will return None, as there is no publisher

# Set 'baz' key, it gets cached

o['baz'] = 'bazzz'

# Now will return cached value, since no provider

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, gr_unittest

class my_add2_dd(gr.feval_dd):

# Boston, MA 02110-1301, USA.

#

import time

import pmt

from gnuradio import gr, gr_unittest, blocks

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr_unittest, blocks, gr, analog

from gnuradio.gr.hier_block2 import _multiple_endpoints, _optional_endpoints

class test_hblk(gr.hier_block2):

def __init__(self, io_sig=1*[gr.sizeof_gr_complex], ndebug=2):

time.sleep(1)

tb.stop()

tb.wait()

def test_012(self):

s, st, h, k = analog.sig_source_c(44100, analog.GR_COS_WAVE, 440, 1.0, 0.0), blocks.message_strobe(pmt.PMT_NIL, 100), blocks.head(gr.sizeof_gr_complex, 1000), test_hblk([gr.sizeof_gr_complex], 16)

tb = gr.top_block()

if __name__ == '__main__':

gr_unittest.run(test_hier_block2, "test_hier_block2.xml")

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, gr_unittest

class test_kludged_imports (gr_unittest.TestCase):

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, gr_unittest

import numpy as np

# Boston, MA 02110-1301, USA.

#

from gnuradio import gr, gr_unittest

class test_tag_utils (gr_unittest.TestCase):

if __name__ == '__main__':

gr_unittest.run(test_tag_utils, "test_tag_utils.xml")

import pmt

class PythonTag(object):

" Python container for tags "

good = False

tag = gr.tag_t()

if(type(tag_struct) == dict):

tag.offset = tag_struct['offset']

good = True

if(isinstance(tag_struct['key'], pmt.swig_int_ptr)):

tag.key = tag_struct['key']

good = True

if(isinstance(tag_struct['value'], pmt.swig_int_ptr)):

tag.value = tag_struct['value']

good = True

if(isinstance(tag_struct['srcid'], pmt.swig_int_ptr)):

tag.srcid = tag_struct['srcid']

good = True

elif(type(tag_struct) == list or type(tag_struct) == tuple):

if(len(tag_struct) == 4):

tag.offset = tag_struct[0]

good = True

good = True

elif(len(tag_struct) == 3):

tag.offset = tag_struct[0]

good = True

# Boston, MA 02110-1301, USA.

#

#import gnuradio.gr.gr_threading as _threading

class _top_block_waiter(_threading.Thread):

"""

"""

GNU radio specific extension of unittest.

"""

import unittest

class TestCase(unittest.TestCase):

"""A subclass of unittest.TestCase that adds additional assertions

as significant digits (measured from the most significant digit).

"""

if round(second.real-first.real, places) != 0:

if round(second.imag-first.imag, places) != 0:

def assertComplexAlmostEqual2 (self, ref, x, abs_eps=1e-12, rel_eps=1e-6, msg=None):

"""

return

if abs(ref) > abs_eps:

else:

def assertComplexTuplesAlmostEqual (self, a, b, places=7, msg=None):

self.assertEqual (len(a), len(b))

self.assertComplexAlmostEqual (a[i], b[i], places, msg)

def assertComplexTuplesAlmostEqual2 (self, ref, x,

abs_eps=1e-12, rel_eps=1e-6, msg=None):

self.assertEqual (len(ref), len(x))

try:

self.assertComplexAlmostEqual2 (ref[i], x[i], abs_eps, rel_eps, msg)

#sys.stderr.write("index = %d " % (i,))

raise

def assertFloatTuplesAlmostEqual (self, a, b, places=7, msg=None):

self.assertEqual (len(a), len(b))

self.assertAlmostEqual (a[i], b[i], places, msg)

def assertFloatTuplesAlmostEqual2 (self, ref, x,

abs_eps=1e-12, rel_eps=1e-6, msg=None):

self.assertEqual (len(ref), len(x))

try:

self.assertComplexAlmostEqual2 (ref[i], x[i], abs_eps, rel_eps, msg)

#sys.stderr.write("index = %d " % (i,))

raise

path = basepath + "/python"

if not os.path.exists(basepath):

xmlrunner = None

# only proceed if .unittests is writable

if(st & stat.S_IWUSR > 0):

# Test if path exists; if not, build it

if not os.path.exists(path):

# Just for safety: make sure we can write here, too

st = os.stat(path)[stat.ST_MODE]

if(st & stat.S_IWUSR > 0):

# Create an XML runner to filename

xmlrunner = gr_xmlrunner.XMLTestRunner(fout)

txtrunner = TextTestRunner(verbosity=1)

suite = TestLoader().loadTestsFromTestCase(PUT)

# use the xmlrunner if we can write the the directory

xmlrunner.run(suite)

main()

"""

XML Test Runner for PyUnit

"""

# Written by Sebastian Rittau <srittau@jroger.in-berlin.de> and placed in

# the Public Domain. With contributions by Paolo Borelli and others.

# Added to GNU Radio Oct. 3, 2010

import os.path

import re

import sys

import time

import unittest

from xml.sax.saxutils import escape

class _TestInfo(object):

supplied stream.

"""

if self._failure is not None:

self._print_error(stream, 'failure', self._failure)

if self._error is not None:

"""Print information from a failure or error to the supplied stream."""

text = escape(str(error[1]))

stream.write('\n')

tb_stream = StringIO()

stream.write(escape(tb_stream.getvalue()))

stream.write(' </%s>\n' % tagname)

stream.write(' ')

output and standard error streams must be passed in.a

"""

for info in self._tests:

info.print_report(stream)

stream.write(' <system-out><![CDATA[%s]]></system-out>\n' % out)

"""Run the given test case or test suite."""

class_ = test.__class__

classname = class_.__module__ + "." + class_.__name__

filename = "TEST-%s.xml" % classname

stream.write('<?xml version="1.0" encoding="utf-8"?>\n')

else:

stream = self._stream

# make this a package

# Import gru stuff

# the Free Software Foundation, Inc., 51 Franklin Street,

# Boston, MA 02110-1301, USA.

#

import os, sys, signal

# Turn application into a background daemon process.

def daemonize(pidfile=None, logfile=None):

# fork() into background

try:

if pid == 0: # First child of first fork()

# Write pid

pid = os.getpid()

if pidfile is not None:

# Redirect streams

if logfile is not None:

# Prevent pinning any filesystem mounts

os.chdir('/')

if __name__ == "__main__":

import time

daemonize()

time.sleep(20)

# DAMAGE.

#

__all__ = ['freqz']

import numpy

y = x * y + p[i]

return y

"""A one-dimensional polynomial class.

p = poly1d([1,2,3]) constructs the polynomial x**2 + 2 x + 3

"""

def __init__(self, c_or_r, r=0):

if isinstance(c_or_r,poly1d):

self.__dict__[key] = c_or_r.__dict__[key]

return

if r:

c_or_r = poly(c_or_r)

c_or_r = atleast_1d(c_or_r)

if len(c_or_r.shape) > 1:

c_or_r = trim_zeros(c_or_r, trim='f')

if len(c_or_r) == 0:

c_or_r = numpy.array([0])

def __pow__(self, val):

if not isscalar(val) or int(val) != val or val < 0:

res = [1]

for k in range(val):

res = polymul(self.coeffs, res)

def __div__(self, other):

if isscalar(other):

else:

other = poly1d(other)

def __rdiv__(self, other):

if isscalar(other):

else:

other = poly1d(other)

def __setattr__(self, key, val):

def __getattr__(self, key):

if key in ['r','roots']:

def __setitem__(self, key, val):

ind = self.order - key

if key < 0:

if key > self.order:

zr = numpy.zeros(key-self.order,self.coeffs.typecode())

self.__dict__['coeffs'] = numpy.concatenate((zr,self.coeffs))

h -- The frequency response.

w -- The frequencies at which h was computed.

"""

if whole:

lastpoint = 2*pi

else:

lastpoint = pi

if worN is None:

N = 512

N = worN

else:

w = worN

w = atleast_1d(w)

zm1 = exp(-1j*w)

# if not plot is None:

# plot(w, h)

return h, w

# Boston, MA 02110-1301, USA.

#

__all__ = ['gnuplot_freqz']

import tempfile

h, w = hw

ampl = 20 * numpy.log10 (numpy.absolute (h) + 1e-9)

if Fs:

for freq, a, ph in zip (w, ampl, phase):

data_file.write ("%g\t%g\t%g\n" % (freq, a, ph))

if __name__ == '__main__':

handle = test_plot ()

#

# Copyright 2005 Free Software Foundation, Inc.

#

#

# Copyright 2005 Free Software Foundation, Inc.

#

#

# Copyright 2005 Free Software Foundation, Inc.

#

return a * b / gcd(a, b)

def log2(x):

To determine if the runner has exited, call exited() on the object.

"""

from gnuradio import gr

import gnuradio.gr.gr_threading as _threading

else:

try:

self._callback(msg)

if self._exit_on_error:

self._exit_error = e

self.stop()

#

# Copyright 2005 Free Software Foundation, Inc.

#

# misc utilities

import types

class seq_with_cursor (object):

__slots__ = [ 'items', 'index' ]

elif initial_index >= 0 and initial_index < len (self.items):

self.index = initial_index

else:

def set_index_by_value(self, v):

"""

cv = self.current()

more = True

while cv < v and more:

new_index = self.index + 1

if new_index < len (self.items):

self.index = new_index

def get_seq (self):

return self.items[:] # copy of items

#

# Copyright 2005 Free Software Foundation, Inc.

#

s = socket.socket (socket.AF_INET, socket.SOCK_STREAM)

try:

s.connect(sock_addr)

sys.stderr.write('Failed to connect to %s: %s\n' %

(sock_addr, os.strerror (err.args[0]),))

sys.exit(1)

s = socket.socket (socket.AF_INET, socket.SOCK_DGRAM)

try:

s.connect(sock_addr)

sys.stderr.write('Failed to connect to %s: %s\n' %

(sock_addr, os.strerror (err.args[0]),))

sys.exit(1)

${CMAKE_BINARY_DIR}/gnuradio-runtime/swig

)

set(GR_TEST_TARGET_DEPS gnuradio-runtime)

endforeach(py_qa_test_file)

endif(ENABLE_TESTING)

vector, dict, uniform_vector, any (boost::any cast)

'''

import os

try:

except ImportError:

dirname, filename = os.path.split(os.path.abspath(__file__))

__path__.append(os.path.join(dirname, "..", "..", "swig"))

# due to changes in the PMT_NIL singleton for static builds, we force

# this into Python here.

PMT_F = get_PMT_F()

PMT_EOF = get_PMT_EOF()

# the Free Software Foundation, Inc., 51 Franklin Street,

# Boston, MA 02110-1301, USA.

import numpy

# SWIG isn't taking in the #define PMT_NIL;

return tuple(elems)

def pmt_from_tuple(p):

return pmt.make_tuple(*args)

def pmt_to_vector(p):

def pmt_from_dict(p):

d = pmt.make_dict()

#dict is immutable -> therefore pmt_dict_add returns the new dict

d = pmt.dict_add(d, python_to_pmt(k), python_to_pmt(v))

return d

def numpy_to_uvector(numpy_array):

try:

mapping = numpy_mappings[numpy_array.dtype]

return mapping[0](numpy_array.size, pc)

except KeyError:

raise ValueError("unsupported numpy array dtype for conversion to pmt %s"%(numpy_array.dtype))

def uvector_to_numpy(uvector):

type_mappings = ( #python type, check pmt type, to python, from python

(None, pmt.is_null, lambda x: None, lambda x: PMT_NIL),

(bool, pmt.is_bool, pmt.to_bool, pmt.from_bool),

(str, pmt.is_symbol, pmt.symbol_to_string, pmt.string_to_symbol),

(int, pmt.is_integer, pmt.to_long, pmt.from_long),

(float, pmt.is_real, pmt.to_double, pmt.from_double),

(complex, pmt.is_complex, pmt.to_complex, pmt.from_complex),

(tuple, pmt.is_tuple, pmt_to_tuple, pmt_from_tuple),

# Boston, MA 02110-1301, USA.

#

import unittest

import pmt

b = pmt.from_double(123765)

d1 = pmt.make_dict()

d2 = pmt.dict_add(d1, a, b)

def test02(self):

const = 123765

# Boston, MA 02110-1301, USA.

#

import unittest

import pmt

class test_pmt_to_python(unittest.TestCase):

def test_pmt_from_double(self):

b = pmt.from_double(123765)

self.assertEqual(pmt.to_python(b), 123765)

def test_numpy_to_uvector_and_reverse(self):

import numpy as np

msg_queue.i

pmt_swig.i

prefs.i

realtime.i

sync_block.i

sync_decimator.i

%template(x_vector_basic_block_sptr) vector<gr::basic_block_sptr>;

};

namespace gr {

class gr::basic_block

}

#ifdef SWIGPYTHON

%pythoncode %{

basic_block_sptr.__repr__ = lambda self: "<basic_block %s (%d)>" % (self.name(), self.unique_id ())

%}

%}

%include <gnuradio/high_res_timer.h>

%include "std_string.i"

%include "stdint.i"

%{

#include <boost/intrusive_ptr.hpp>

#include <boost/shared_ptr.hpp>

%template(pmt_vector_cfloat) std::vector< std::complex<float> >;

%template(pmt_vector_cdouble) std::vector< std::complex<double> >;

////////////////////////////////////////////////////////////////////////

// Language independent exception handler

////////////////////////////////////////////////////////////////////////