rayleigh_num_bins = 31

laplace_num_bins = 31

#*** GENERATE DATA ***#

#*** HISTOGRAM DATA AND CALCULATE EXPECTED COUNTS ***#

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

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

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

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

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

plt.figure(1)

plt.xlabel('Bins'), plt.ylabel('Count'), plt.title('Uniform: Distribution')

plt.figure(2)

plt.xlabel('Bins'), plt.ylabel('Count'), plt.title('Gauss: Distribution')

plt.figure(3)

plt.xlabel('Bins'), plt.ylabel('Count'), plt.title('Rayleigh: Distribution')

plt.figure(4)

plt.xlabel('Bins'), plt.ylabel('Count'), plt.title('Laplace: Distribution')

plt.show()

from argparse import ArgumentParser

import sys

class affinity_set(gr.top_block):

def __init__(self):

# Blocks

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

vec_len = 1

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

# Connections

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

self.connect((self.blocks_throttle_0, 0), (self.filter_filt_0, 0))

self.connect((self.filter_filt_0, 0), (self.filter_filt_1, 0))

self.connect((self.filter_filt_1, 0), (self.blocks_null_sink_0, 0))

def set_samp_rate(self, samp_rate):

self.samp_rate = samp_rate

if __name__ == '__main__':

parser = ArgumentParser()

args = parser.parse_args()

except ValueError:

print("Invalid number")

else:

def handle_file(input_filename):

cmd_file_name = cmd_file.name

data_file_name = data_file.name

desc = parse_file(input_filename, data_file)

if len(desc) > 0:

os.system("gnuplot " + cmd_file_name + " -")

def main():

# 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)

f.write(cmd)

f.write('if test $? -ge 128; then exit 128; fi\n')

f.flush()

def main():

description = """%prog gathers multiprocessor scaling data using the ./synthetic.py benchmark.

All combinations of npipes and nstages between 1 and --max-pipes-and-stages are tried.

"""

parser = ArgumentParser(description=description)

parser.add_argument("-d", "--description", metavar="DESC",

parser.add_argument("-n", "--ncores", type=int, default=1,

parser.add_argument("-g", "--gflops", metavar="GFLOPS", type=float, default=3.0,

parser.add_argument("-m", "--max-pipes-and-stages", metavar="MAX", type=int, default=16,

args = parser.parse_args()

shell = os.popen("/bin/sh", "w")

args.gflops,

args.max_pipes_and_stages)

if __name__ == '__main__':

main()

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)

default_nsamples = 10e6

parser = ArgumentParser()

parser.add_argument("-p", "--npipelines", type=intx, default=1,

parser.add_argument("-s", "--nstages", type=intx, default=1, metavar="NSTAGES",

parser.add_argument("-N", "--nsamples", type=eng_float, default=default_nsamples,

parser.add_argument("-m", "--machine-readable", action="store_true", default=False,

args = parser.parse_args()

start = os.times()

tb.run()

stop = os.times()

user, sys, childrens_user, childrens_sys, real = delta

total_user = user + childrens_user

if tb.machine_readable:

print("%3d %3d %.3e %7.3f %7.3f %7.3f %7.3f %.6e %.3e" % (

else:

print("real %7.3f" % (real,))

print("user %7.3f" % (total_user,))

print("sys %7.3f" % (total_sys,))

print("(user+sys)/real %7.3f" % ((total_user + total_sys) / real,))

if __name__ == "__main__":

time_it(tb)

except KeyboardInterrupt:

raise SystemExit(128)

import sys

import math

class wfm_rx_block (gr.top_block):

def __init__(self):

gr.top_block.__init__(self)

parser = ArgumentParser(description="Decode WFM signal into WAV file.")

parser.add_argument("-V", "--volume", type=eng_float,

parser.add_argument("input_file", help="Input file (complex samples)")

parser.add_argument("output_file", help="Output WAV file")

# build graph

adc_rate = 64e6 # 64 MS/s

usrp_decim = 200

audio_decimation = 10

audio_rate = demod_rate / audio_decimation # 32 kHz

#self.guts = analog.wfm_rcv (demod_rate, audio_decimation)

# FIXME rework {add,multiply}_const_* to handle multiple streams

self.volume_control_l = blocks.multiply_const_ff(self.vol)

sink = blocks.wavfile_sink(args.output_file, 2, int(audio_rate),

blocks.FORMAT_WAV, blocks.FORMAT_PCM_16)

else:

# now wire it all together

if args.volume is None:

g = self.volume_range()

# set initial values

self.set_vol(args.volume)

g = self.volume_range()

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

self.volume_control_l.set_k(10**(self.vol / 10))

try:

from gnuradio import audio

except ImportError:

sys.exit(1)

class audio_sink(gr.top_block):

def __init__(self, host, port, pkt_size, sample_rate, eof):

gr.top_block.__init__(self, "audio_sink")

dst = audio.sink(sample_rate)

self.connect(src, dst)

if __name__ == '__main__':

parser = ArgumentParser()

parser.add_argument("--host", default="0.0.0.0",

parser.add_argument("--port", type=int, default=65500,

parser.add_argument("--packet-size", type=int, default=1472,

parser.add_argument("-r", "--sample-rate", type=int, default=32000,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = audio_sink(args.host, args.port,

except KeyboardInterrupt:

# Ctrl-C exits

pass

try:

from gnuradio import audio

except ImportError:

sys.exit(1)

class audio_source(gr.top_block):

def __init__(self, host, port, pkt_size, sample_rate, eof):

gr.top_block.__init__(self, "audio_source")

self.audio = audio.source(sample_rate)

self.connect(self.audio, self.sink)

if __name__ == '__main__':

parser = ArgumentParser()

parser.add_argument("--host", default="127.0.0.1",

parser.add_argument("--port", type=int, default=65500,

parser.add_argument("--packet-size", type=int, default=1472,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = audio_source(args.host, args.port,

except KeyboardInterrupt:

# Ctrl-C exits

pass

from gnuradio import blocks

from argparse import ArgumentParser

class dial_tone_sink(gr.top_block):

def __init__(self, host, port, pkt_size, sample_rate, eof):

gr.top_block.__init__(self, "dial_tone_sink")

sink = audio.sink(sample_rate)

self.connect(udp, sink)

if __name__ == '__main__':

parser = ArgumentParser()

parser.add_argument("--host", default="0.0.0.0",

parser.add_argument("--port", type=int, default=65500,

parser.add_argument("--packet-size", type=int, default=1472,

parser.add_argument("-r", "--sample-rate", type=int, default=8000,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = dial_tone_sink(args.host, args.port,

except KeyboardInterrupt:

# Ctrl-C exits

pass

except ImportError:

sys.stderr.write("This example requires gr-blocks.\n")

class dial_tone_source(gr.top_block):

def __init__(self, host, port, pkt_size, sample_rate, eof):

gr.top_block.__init__(self, "dial_tone_source")

amplitude = 0.3

add = blocks.add_ff()

# Throttle needed here to account for the other side's audio card sampling rate

self.connect(src1, (add, 1))

self.connect(add, thr, sink)

if __name__ == '__main__':

parser = ArgumentParser()

parser.add_argument("--host", default="127.0.0.1",

parser.add_argument("--port", type=int, default=65500,

parser.add_argument("--packet-size", type=int, default=1472,

parser.add_argument("-r", "--sample-rate", type=int, default=8000,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = dial_tone_source(args.host, args.port,

from gnuradio.eng_arg import eng_float, intx

from argparse import ArgumentParser

class vector_sink(gr.top_block):

def __init__(self, host, port, pkt_size, eof):

gr.top_block.__init__(self, "vector_sink")

sink = blocks.file_sink(gr.sizeof_float, "received.dat")

self.connect(udp, sink)

if __name__ == "__main__":

parser = ArgumentParser()

parser.add_argument("-H", "--host", default="0.0.0.0",

parser.add_argument("-p", "--port", type=int, default=65500,

parser.add_argument("-s", "--packet-size", type=int, default=1471,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = vector_sink(args.host, args.port,

except KeyboardInterrupt:

# Ctrl-C exits

pass

from gnuradio import blocks

from argparse import ArgumentParser

class vector_source(gr.top_block):

def __init__(self, host, port, pkt_size, eof):

gr.top_block.__init__(self, "vector_source")

vec = blocks.vector_source_f(data, True)

udp = blocks.udp_sink(gr.sizeof_float, host, port, pkt_size, eof=eof)

self.connect(vec, udp)

if __name__ == '__main__':

parser = ArgumentParser()

parser.add_argument("--host", default="127.0.0.1",

parser.add_argument("--port", type=int, default=65500,

parser.add_argument("--packet-size", type=int, default=1471,

parser.add_argument("--no-eof", action="store_true", default=False,

args = parser.parse_args()

# Create an instance of a hierarchical block

top_block = vector_source(args.host, args.port, args.packet_size,

except KeyboardInterrupt:

# Ctrl-C exits

pass

import math

import sys

"""return a list of nentries containing tuples of the form:

(m, c). min_x and max_x specify the domain

of the table.

"""

r = []

a = (i * incx) + min_x

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

c = f(a)

return r

nbits = 10

nentries = 2**nbits

# min_x = -2**31

# max_x = 2**31-1

min_x = 0

# sys.stdout.write ('static const int WORDBITS = 32;\n')

# sys.stdout.write ('static const int NBITS = %d;\n' % (nbits,))

# sys.stdout.write ('static const double sine_table[%d][2] = {\n'% (nentries,))

for e in t:

# sys.stdout.write ('};\n')

if __name__ == '__main__':

# https://docs.python.org/3/whatsnew/3.8.html#bpo-36085-whatsnew

if os.name == 'nt' and hasattr(os, 'add_dll_directory'):

root_dir = __file__

root_dir = os.path.dirname(root_dir)

bin_dir = os.path.join(root_dir, 'bin')

if os.path.exists(bin_dir):

except Exception as ex:

break

# Check if the gnuradio package is installed or whether we're attempting to import it from

# the build directory.

path = os.path.abspath(__file__)

if path.endswith('.pyc'):

path = path[:-1]

"""

class GNURadioControlPortClient(object):

"""

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

such as QtGui.QApplication.exec_

"""

def __init__(self, host=None, port=None, rpcmethod='thrift', callback=None, blockingcallback=None):

self.client = None

from gnuradio import blocks

from gnuradio import filter, fft

from gnuradio.ctrlport.GNURadio import ControlPort

try:

from gnuradio import qtgui

print("Error: Program requires PyQt5 and gr-qtgui.")

sys.exit(1)

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

# Setup signals

plotupdated = QtCore.pyqtSignal(Qt.QWidget)

self._name = name

self._npts = 500

self._rate = rate

self.layout = Qt.QVBoxLayout()

self.setLayout(self.layout)

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

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

for n in range(1, self._ncons):

self._ncons = nconnections

self.thr = blocks.throttle(self._datasize, self._rate)

self.snk = self.get_qtsink()

for n in range(self._ncons):

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

self.src.append(self.get_vecsource())

if(n == 0):

self.connect(self.src[n], self.thr)

else:

self.py_window = sip.wrapinstance(self.snk.qwidget(), Qt.QWidget)

if(self._iscomplex):

data_r = data[0::2]

data_i = data[1::2]

return data

def update(self, data):

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

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

else:

self._data_len[n] = npts

else:

self._npts = npts

self.snk.reset()

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

self._last_data[n] = data[n][-self._npts:]

else:

newdata += data[n]

self.src[n].set_data(newdata)

self._last_data[n] = newdata

if(self._iscomplex):

data[n] = complex(data[n][0], data[n][1])

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

else:

for n in range(self._ncons):

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

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

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

class GrDataPlotterC(GrDataPlotParent):

def __init__(self, name, rate, pmin=None, pmax=None, stripchart=False):

GrDataPlotParent.__init__(self, name, rate, pmin, pmax)

return self._npts

def set_line_label(self, n, name):

class GrDataPlotterF(GrDataPlotParent):

def set_line_label(self, n, name):

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

class GrTimeRasterB(GrDataPlotParent):

def __init__(self, name, rate, pmin=None, pmax=None):

GrDataPlotParent.__init__(self, name, rate, pmin, pmax)

self.uid = uid

self.treeWidget = Qt.QTreeWidget(parent)

self.treeWidget.setColumnCount(len(headers))

self.treeWidget.setHeaderLabels(headers)

self.treeWidget.resizeColumnToContents(0)

descr = str(knobprops[itemKey].description)

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

# If it's a byte stream, Python thinks it's a string.

# Unpack and convert to floats for plotting.

# Ignore the edge list knob if it's being exported

elif(type(v) == str and itemKey.find('probe2_b') == 0):

# Convert the final value to a string for displaying

v = str(v)

if (item.text(1) != v or

item.text(2) != units or

item.setText(1, v)

item.setText(2, units)

if k not in foundKeys:

v = knobs[k].value

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

# If it's a byte stream, Python thinks it's a string.

# Unpack and convert to floats for plotting.

# Ignore the edge list knob if it's being exported

elif(type(v) == str and k.find('probe2_b') == 0):

item = Qt.QTreeWidgetItem([k, str(v),

self.treeWidget.addTopLevelItem(item)

# Remove items currently in tree that are not in the knob list.

for itemKey in deleteKeys:

if (len(qtwiList) > 1):

raise Exception('More than one item with key %s in tree' %

itemKey)

RPCMethods = {'thrift': 'Apache Thrift',

"""

host: hostname of the connection

"""

class RPCConnection(object):

def __init__(self, method, port, host=None):

(self.method, self.port) = (method, port)

def __str__(self):

def getName(self):

return RPCMethods[self.method]

def getKnobs(self, *args):

raise NotImplementedError()

raise NotImplementedError()

raise NotImplementedError()

raise NotImplementedError()

def shutdown(self):

import pmt

import sys

class ThriftRadioClient(object):

def __init__(self, host, port):

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

def getRadio(self):

return self.radio

"""

RPC Client interface for the Apache Thrift middle-ware RPC transport.

host: hostname of the connection

"""

class RPCConnectionThrift(RPCConnection.RPCConnection):

class Knob(object):

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

else:

port = int(port)

self.newConnection(host, port)

self.unpack_dict = {

}

self.pack_dict = {

}

def __str__(self):

if(f):

return f(key, knob)

else:

raise exceptions.ValueError

def packKnob(self, knob):

if(f):

return f(knob)

else:

raise exceptions.ValueError

def newConnection(self, host=None, port=None):

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

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

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

return knobprops

def getKnobs(self, *args):

result[key] = knob

return result

result = {}

for key, knob in list(self.thriftclient.radio.getRe(*args).items()):

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

result[k.key] = self.packKnob(k)

self.thriftclient.radio.setKnobs(result)

else:

def shutdown(self):

self.thriftclient.radio.shutdown()

serialized. The msg is already a PMT and so just serialized.

'''

self.thriftclient.radio.postMessage(pmt.serialize_str(pmt.intern(blk_alias)),

pmt.serialize_str(msg))

def printProperties(self, props):

info = ""

info += "Item:\t\t{0}\n".format(props.description)

from PyQt5 import QtCore, Qt

from argparse import ArgumentParser

from gnuradio import gr, ctrlport

from gnuradio.ctrlport.GrDataPlotter import *

from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient

class RateDialog(Qt.QDialog):

def __init__(self, delay, parent=None):

super(RateDialog, self).__init__(parent)

self.setWindowTitle("Update Delay (ms)")

self.delay = Qt.QLineEdit(self)

self.delay.setText(str(delay))

self.gridLayout.addWidget(self.delay)

self.gridLayout.addWidget(self.buttonBox)

self.buttonBox.accepted.connect(self.accept)

self.buttonBox.rejected.connect(self.reject)

def accept(self):

self.done(1)

def reject(self):

self.done(0)

class MAINWindow(Qt.QMainWindow):

def minimumSizeHint(self):

def __init__(self, radioclient):

self.setUnifiedTitleAndToolBarOnMac(True)

self.newCon(radioclient)

self.setWindowIcon(icon)

# Locally turn off ControlPort export from GR. This prevents

# been specified).

os.environ['GR_CONF_CONTROLPORT_ON'] = 'False'

self.updateRate = int(nur)

for c in self.conns:

c.updateRate = self.updateRate

c.timer.setInterval(self.updateRate)

def newCon(self, radioclient=None):

if(child.radioclient is not None):

child.setWindowTitle(str(child.radioclient))

self.mdiArea.addSubWindow(child)

props = radio.properties([key])

# Use display option mask of item to set up available plot

# types and default options.

knobprop = self.knobprops[uid][tag]

strr = str(tree.radioclient)

disp = knobprop.display

if(disp & gr.DISPTIME):

self.mdiArea.addSubWindow(plot)

plot.setWindowTitle("{0}: {1}".format(title, plot.name()))

# when the plot is updated via drag-and-drop, we need to be

# notified of the new qwidget that's created so we can

for i, plots in enumerate(self.plots):

for p in plots:

if(p == q):

q.qwidget().destroyed.connect(lambda obj=None, plot=p: self.destroyPlot(plot=plot))

break

# If it's a byte stream, Python thinks it's a string.

# Unpack and convert to floats for plotting.

if(type(d) == str and n.find('probe2_b') == 0):

d = [float(di) for di in d]

data.append(d)

if window:

self.mdiArea.setActiveSubWindow(window)

def createActions(self):

self.newConAct = Qt.QAction("&New Connection",

self.exitAct = Qt.QAction("E&xit", self, shortcut="Ctrl+Q",

self.closeAct = Qt.QAction("Cl&ose", self, shortcut="Ctrl+F4",

self.closeAllAct = Qt.QAction("Close &All", self,

self.urAct = Qt.QAction("Update Rate", self, shortcut="F5",

qks = Qt.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.Key_T)

self.tileAct = Qt.QAction("&Tile", self,

qks = Qt.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.Key_C)

self.cascadeAct = Qt.QAction("&Cascade", self,

self.nextAct = Qt.QAction("Ne&xt", self,

self.previousAct = Qt.QAction("Pre&vious", self,

self.separatorAct = Qt.QAction(self)

self.separatorAct.setSeparator(True)

self.aboutAct = Qt.QAction("&About", self,

self.aboutQtAct = Qt.QAction("About &Qt", self,

def createMenus(self):

self.fileMenu = self.menuBar().addMenu("&File")

def createStatusBar(self):

self.statusBar().showMessage("Ready")

def activeMdiChild(self):

activeSubWindow = self.mdiArea.activeSubWindow()

if activeSubWindow:

def about(self):

about_info = \

This program is part of GNU Radio.\n

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.\n

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.\n

self.gridLayout = Qt.QGridLayout(self)

self.host = Qt.QLineEdit(self)

self.port = Qt.QLineEdit(self)

self.host.setText("localhost")

self.port.setText("43243")

self.gridLayout.addWidget(self.host)

self.gridLayout.addWidget(self.port)

self.buttonBox.accepted.connect(self.accept)

self.buttonBox.rejected.connect(self.reject)

def accept(self):

self.done(1)

self.key = key

self.radio = radio

self.layout = Qt.QVBoxLayout()

self.props = radio.properties([key])[key]

except:

has_set = False

if(has_set is False):

self.cancelButton = Qt.QPushButton("Ok")

self.cancelButton.clicked.connect(self.reject)

self.buttonlayout.addWidget(self.cancelButton)

self.layout.addLayout(self.buttonlayout)

self.textInput = Qt.QLineEdit()

self.layout.addWidget(self.textInput)

rv = radio.getKnobs([key])

val = rv[key].value

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

self.textInput.setText(str(val))

self.sv = rv[key]

self.setButton.clicked.connect(self._set)

self.cancelButton.clicked.connect(self.reject)

if(self.is_num):

self.sliderlayout = Qt.QHBoxLayout()

self.slider = Qt.QSlider(QtCore.Qt.Horizontal)

self.sliderlayout.addWidget(self.slider)

self.steps = 10000

self.valspan = self.props.max.value - self.props.min.value

self.setLayout(self.layout)

def _set_slider_value(self, val):

def _slide(self):

self.textInput.setText(str(val))

def _apply(self):

val.re = t.real

val.im = t.imag

else:

return

self.sv.value = val

st = time.time()

knobs = self.radioclient.getKnobs([])

ft = time.time()

except Exception as e:

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

# Find window of connection

remove = []

tableitems = knobs.keys()

self.table.updateItems(knobs, self.knobprops)

self.parent.update(knobs, self.uid)

def __init__(self, radioclient, uid=0, updateRate=2000, parent=None):

super(MForm, self).__init__()

port = str(askinfo.port.text())

try:

print("Connected to %s:%s" % (host, port))

except:

print("Cannot connect to %s:%s" % (host, port))

self.knobprops = self.radioclient.properties([])

self.parent.knobprops.append(self.knobprops)

self.timer = QtCore.QTimer()

self.constupdatediv = 0

self.tableupdatediv = 0

# make table

self.table = GrDataPlotterValueTable(uid, self, 0, 0, 400, 200)

self.table.treeWidget.setSizePolicy(sizePolicy)

self.table.treeWidget.setSortingEnabled(True)

self.table.treeWidget.setDragEnabled(True)

if pmin == []:

pmin = None

else:

if pmax == []:

pmax = None

else:

return pmin, pmax

class MyApp(object):

def __init__(self, args):

parser = ArgumentParser(description="GNU Radio Control Port Monitor")

parser.add_argument("port", help="port")

args = parser.parse_args()

try:

except:

print("ControlPort failed to connect. Check the config of your endpoint.")

print("\t[ControlPort] on = True")

def run(self, client):

MAINWindow(client).show()

MyApp(sys.argv)

from argparse import ArgumentParser

from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient

print(sys.argv[0], "could not load QTAgg backend.")

sys.exit(1)

from matplotlib.figure import Figure

except ImportError:

print(sys.argv[0], "requires networkx and matplotlib.",

sys.exit(1)

from PyQt5 import QtCore, Qt

from networkx.drawing.nx_agraph import graphviz_layout

class MAINWindow(Qt.QMainWindow):

def minimumSizeHint(self):

def __init__(self, radioclient):

self.newCon(radioclient)

self.setWindowIcon(icon)

def newSubWindow(self, window, title):

if window:

self.mdiArea.setActiveSubWindow(window)

def createActions(self):

self.newConAct = Qt.QAction("&New Connection",

self.exitAct = Qt.QAction("E&xit", self, shortcut="Ctrl+Q",

self.closeAct = Qt.QAction("Cl&ose", self, shortcut="Ctrl+F4",

self.closeAllAct = Qt.QAction("Close &All", self,

qks = Qt.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.Key_T)

self.tileAct = Qt.QAction("&Tile", self,

qks = Qt.QKeySequence(QtCore.Qt.CTRL + QtCore.Qt.Key_C)

self.cascadeAct = Qt.QAction("&Cascade", self,

self.nextAct = Qt.QAction("Ne&xt", self,

self.previousAct = Qt.QAction("Pre&vious", self,

self.separatorAct = Qt.QAction(self)

self.separatorAct.setSeparator(True)

self.aboutAct = Qt.QAction("&About", self,

self.aboutQtAct = Qt.QAction("About &Qt", self,

def createMenus(self):

self.fileMenu = self.menuBar().addMenu("&File")

def createStatusBar(self):

self.statusBar().showMessage("Ready")

def activeMdiChild(self):

activeSubWindow = self.mdiArea.activeSubWindow()

if activeSubWindow:

def about(self):

about_info = \

This program is part of GNU Radio.\n

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.\n

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.\n

self.gridLayout = Qt.QGridLayout(self)

self.host = Qt.QLineEdit(self)

self.port = Qt.QLineEdit(self)

self.host.setText("localhost")

self.port.setText("43243")

self.buttonBox = Qt.QDialogButtonBox(Qt.QDialogButtonBox.Ok |

self.gridLayout.addWidget(self.host)

self.gridLayout.addWidget(self.port)

self.buttonBox.accepted.connect(self.accept)

self.buttonBox.rejected.connect(self.reject)

def accept(self):

self.done(1)

self.perfTable = Qt.QTableWidget()

self.perfTable.setColumnCount(2)

self.perfTable.verticalHeader().hide()

self.perfTable.horizontalHeader().setStretchLastSection(True)

self.perfTable.setSortingEnabled(True)

nodes = self.G.nodes(data=True)

# set up plot

self.sp = self.f.add_subplot(111)

self.sp.set_autoscale_on(True)

# set up tabs

self.tabber = Qt.QTabWidget()

self.layout.addWidget(self.tabber)

self.tabber.addTab(self.f.canvas, "Graph View")

# set up timer

for i, node in enumerate(nodes):

self.perfTable.setItem(i, 0,

for k in data.keys():

weight = data[k]

if(len(existing) == 0):

i = self.perfTable.rowCount()

else:

def stat_changed(self, index):

self._statistic = str(self.stattype.currentText())

def checksort_changed(self, state):

self._sort = state > 0

class DataTableBuffers(DataTable):

def __init__(self, radioclient, G):

super(DataTableBuffers, self).__init__(radioclient, G)

def update(self):

nodes = self.G.nodes()

# get buffer fullness for all blocks

try:

buf_knobs = self.radioclient.getKnobs(kl)

except Exception as e:

self.parentWidget().mdiArea().removeSubWindow(self.parentWidget())

self.close()

return

for blk in buffer_fullness:

bdata = buffer_fullness[blk]

if bdata:

if(self.perfTable.isVisible()):

self.table_update(blockport_fullness)

if(self._sort):

sorted_fullness = sorted(blockport_fullness.items(),

key=operator.itemgetter(1))

else:

if self._keymap:

for b in blockport_fullness:

else:

sorted_fullness = blockport_fullness.items()

if(not self.disp):

alpha=0.5, align='edge')

self.sp.set_ylabel("% Buffers Full")

self.sp.set_xticklabels(list(map(lambda x: " " + x, map(lambda x: x[0], sorted_fullness))),

rotation="vertical", verticalalignment="bottom")

else:

self.sp.set_xticklabels(list(map(lambda x: " " + x, map(lambda x: x[0], sorted_fullness))),

rotation="vertical", verticalalignment="bottom")

r.set_height(w[1])

self.f.canvas.draw()

class DataTableRuntimes(DataTable):

def __init__(self, radioclient, G):

def update(self):

nodes = self.G.nodes()

# get work time for all blocks

try:

wrk_knobs = self.radioclient.getKnobs(kl)

except Exception as e:

self.parentWidget().mdiArea().removeSubWindow(self.parentWidget())

self.close()

return

total_work = 1

work_times = dict(zip(

map(lambda x: x.split("::")[0], wrk_knobs.keys()),

# update table view

if(self.perfTable.isVisible()):

else:

if(self._sort):

self._keymap = list(map(operator.itemgetter(0), sorted_work))

else:

if self._keymap:

for b in work_times:

sorted_work[self._keymap.index(b)] = (b, work_times[b])

else:

sorted_work = work_times.items()

if(not self.disp):

list(map(lambda x: x[1], sorted_work)),

alpha=0.5, align='edge')

self.sp.set_ylabel("% Runtime")

self.sp.set_xticklabels(list(map(lambda x: " " + x[0], sorted_work)),

else:

self.sp.set_xticklabels(list(map(lambda x: " " + x[0], sorted_work)),

r.set_height(w[1])

self.f.canvas.draw()

class MForm(Qt.QWidget):

def update(self):

try:

kl1 = self.radioclient.getKnobs([self.clockKey])

self.clockKey = list(kl1.keys())[0]

self.currClock = kl1[self.clockKey].value

self.clockSel.setCurrentIndex(self.clockSelIdx)

self.prevent_clock_change = False

except Exception as e:

nodes_msg = self.G_msg.nodes()

# get current buffer depths of all output buffers

st = time.time()

buf_knobs = self.radioclient.getKnobs(kl)

map(lambda x: x.value, buf_knobs.values())))

# get work time for all blocks

st = time.time()

wrk_knobs = self.radioclient.getKnobs(kl)

td2 = time.time() - st

if(total_work == 0):

total_work = 1

work_times = dict(zip(

work_times_padded = dict(zip(

work_times_padded.update(work_times)

for n in nodes_stream:

# ne is the list of edges away from this node!

# get the right output buffer/port weight for each edge

sourceport = e[2]["sourceport"]

if(e[2]["type"] == "stream"):

e[2]["weight"] = newweight

for n in nodes_msg:

sourceport = e[2]["sourceport"]

if(e[2]["type"] == "msg"):

newweight = 0.01

# set updated weights

#self.node_weights = map(lambda x: 20+2000*work_times[x], nodes_stream)

# draw graph updates

if(self.do_update):

self.updateGraph()

latency = td1 + td2

except Exception as e:

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

# Find window of connection

for p in self.parent.mdiArea.subWindowList():

return

def rtt(self):

def bpt(self):

def resetPCs(self):

knobs = []

for b in self.blocks_list:

k = self.radioclient.setKnobs(knobs)

def toggleFlowgraph(self):

host = str(askinfo.host.text())

port = str(askinfo.port.text())

try:

print("Connected to %s:%s" % (host, port))

except:

print("Error connecting to %s:%s" % (host, port))

else:

self.radioclient = radioclient

if self.radioclient is None:

return

self.layoutTop.addLayout(self.layout)

self.rttAct = Qt.QAction("Runtime Table",

self.rttBut = Qt.QToolButton()

self.rttBut.setDefaultAction(self.rttAct)

self.ctlBox.addWidget(self.rttBut)

self.bptAct = Qt.QAction("Buffer Table",

self.bptBut = Qt.QToolButton()

self.bptBut.setDefaultAction(self.bptAct)

self.ctlBox.addWidget(self.bptBut)

self.resetPCsAct = Qt.QAction("Reset", self,

self.resetPCsBut = Qt.QToolButton()

self.resetPCsBut.setDefaultAction(self.resetPCsAct)

self.ctlBox.addWidget(self.resetPCsBut)

self.pauseFGAct = Qt.QAction("Pause", self,

self.pauseFGAct.setCheckable(True)

self.pauseFGBut = Qt.QToolButton()

self.pauseFGBut.setDefaultAction(self.pauseFGAct)

self.prevent_clock_change = True

self.clockKey = None

self.clockSel = Qt.QComboBox(self)

list(map(lambda x: self.clockSel.addItem(x), self.clocks.keys()))

self.ctlBox.addWidget(self.clockSel)

self.timer = QtCore.QTimer()

self.constupdatediv = 0

self.tableupdatediv = 0

# Set up the graph of blocks

tmplist = []

knobs = self.radio.getKnobs([])

if(input_name(blockname) in knobs):

tmplist.append(blockname)

if not edgelist:

sys.exit(1)

self.blocks_list = tmplist

# add stream connections

for e in edges:

_e = e.split("->")

# add msg connections

for e in msgedges:

_e = e.split("->")

self.G = nx.MultiDiGraph()

self.G_stream = nx.MultiDiGraph()

n_edges = self.G.edges(data=True)

for e in n_edges:

self.G = nx.MultiDiGraph()

self.G.add_edges_from(n_edges)

self.sp = self.f.add_subplot(111)

self.sp.set_autoscale_on(True)

self.layout.addWidget(self.f.canvas)

thrsh = 100

if(x is not None and y is not None):

i = nearby.index(min(nearby))

if(abs(list(self.pos.values())[i][0] - x) < thrsh and

self._current_block = list(self.pos.keys())[i]

self._grabbed = True

def mouse_move(self, event):

if self._grabbed:

x, y = event.xdata, event.ydata

if(x is not None and y is not None):

self.updateGraph()

def button_release(self, event):

self._grabbed = False

def openMenu(self, pos):

index = self.table.treeWidget.selectedIndexes()

item = self.table.treeWidget.itemFromIndex(index[0])

nx.draw(self.G, self.pos,

edge_color=self.edge_weights,

node_color='#A0CBE2',

node_shape="s",

node_size=self.node_weights,

edge_cmap=plt.cm.Reds,

ax=self.sp,

arrows=False

nx.draw_networkx_labels(self.G, self.pos,

font_size=12)

nx.draw_networkx_edges(self.G, self.pos,

edge_color=self.edge_weights,

edge_cmap=plt.cm.Reds,

ax=self.sp,

arrows=False)

def __init__(self, args):

parser = ArgumentParser(description="GNU Radio Performance Monitor")

parser.add_argument("port", help="port")

args = parser.parse_args()

signal.signal(signal.SIGTERM, signal.SIG_DFL)

try:

except:

print("ControlPort failed to connect. Check the config of your endpoint.")

print("\t[ControlPort] on = True")

def run(self, client):

MAINWindow(client).show()

MyApp(sys.argv)

#

from gnuradio import gr

class monitor(object):

print("ControlPort Monitor running.")

self.started = False

self.tool = tool

atexit.register(self.shutdown)

# setup export prefs

if(tool == "gr-perf-monitorx"):

def start(self):

try:

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

self.proc = subprocess.Popen(cmd)

self.started = True

except:

import argparse

from gnuradio import eng_notation

def intx(string):

"""

Generic integer type, will interpret string as string literal.

"Invalid integer value: {}".format(string)

)

def eng_float(string):

"""

Takes a string, returns a float. Accepts engineering notation.

raise argparse.ArgumentTypeError(

"Invalid engineering notation value: {}".format(string)

)

scale_factor['f'] = 1e-15

scale_factor['a'] = 1e-18

'''Convert a number to a string in engineering notation. E.g., 5e-9 -> 5n'''

m = abs(n)

format_spec = '%.' + repr(int(precision)) + 'g'

return '%s' % float(format_spec % (n))

'''Convert a string in engineering notation to a number. E.g., '15m' -> 15e-3'''

try:

if not isinstance(value, str):

scale = 1.0

suffix = value[-1]

if suffix in scale_factor:

except (TypeError, KeyError, ValueError):

"Invalid engineering notation value: %r" % (value,))

from optparse import Option, OptionValueError

from . import eng_notation

try:

return eng_notation.str_to_num(value)

except (ValueError, TypeError):

"option %s: invalid engineering notation value: %r" % (opt, value))

try:

except (ValueError, TypeError):

"option %s: invalid integer value: %r" % (opt, value))

class eng_option (Option):

TYPES = Option.TYPES + ("eng_float", "intx", "subdev")

TYPE_CHECKER["eng_float"] = check_eng_float

TYPE_CHECKER["intx"] = check_intx

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

# Otherwise it will reside in bindings/.

try:

from .gr_python import *

"""Not a directed acyclic graph"""

pass

class CantHappen (Exception):

"""Can't happen"""

pass

#

import numpy

import ctypes

return ()

if nports <= ntypes:

return self.__types[:nports]

def __iter__(self):

"""

# Normalize the many Python ways of saying 'nothing' to '()'

in_sig = in_sig or ()

out_sig = out_sig or ()

# Backward compatibility: array of type strings -> py_io_signature

if type(in_sig) is py_io_signature:

self.__in_sig = in_sig

Makes this block connectable by hier/top block python

"""

return self.gateway.to_basic_block()

def fixed_rate_noutput_to_ninput(self, noutput_items):

return int((noutput_items * self._decim / self._interp) + self.gateway.history() - 1)

def handle_forecast(self, noutput_items, ninputs):

"""

block_gateway in c++ across pybind11 wrappers

"""

return self.forecast(noutput_items, ninputs)

# return ninput_items_required

def forecast(self, noutput_items, ninputs):

forecast is only called from a general block

this is the default implementation

"""

for i in range(ninputs):

return ninput_items_required

ninputs = len(input_items)

noutputs = len(output_items)

in_types = self.in_sig().port_types(ninputs)

out_types = self.out_sig().port_types(noutputs)

ctypes.pythonapi.PyCapsule_GetPointer.restype = ctypes.c_void_p

if self._block_type != gr.GW_BLOCK_GENERAL:

in_types[i],

self.fixed_rate_noutput_to_ninput(noutput_items)

) for i in range(ninputs)]

else:

in_types[i],

ninput_items[i]

) for i in range(ninputs)]

out_types[i],

noutput_items

if self._block_type != gr.GW_BLOCK_GENERAL:

self.consume_items(r)

else:

return r

return self.__out_sig

def set_msg_handler(self, which_port, handler_function):

# Save handler object in class so it's not garbage collected

self.msg_handlers[which_port] = handler_function

# return self.gateway.produce(which_output, how_many_items)

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

# Wrappers for the user to inherit from

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

"""

Args:

name (str): block name

For backward compatibility, a sequence of numpy type names is also

accepted as an io signature.

"""

def __init__(self, name, in_sig, out_sig):

gateway_block.__init__(self,

name=name,

block_type=gr.GW_BLOCK_GENERAL

)

pass

"""

Args:

name (str): block name

For backward compatibility, a sequence of numpy type names is also

accepted as an io signature.

"""

def __init__(self, name, in_sig, out_sig):

gateway_block.__init__(self,

name=name,

self._decim = 1

self._interp = 1

if (nitems > 0):

self.gateway.consume_each(nitems)

For backward compatibility, a sequence of numpy type names is also

accepted as an io signature.

"""

def __init__(self, name, in_sig, out_sig, decim):

gateway_block.__init__(self,

name=name,

self.gateway.set_relative_rate(self._interp, self._decim)

self.gateway.set_output_multiple(self._interp)

if (nitems > 0):

self.gateway.consume_each(int(nitems * self._decim))

return [self.fixed_rate_noutput_to_ninput(noutput_items) for ii in range(ninputs)]

class interp_block(gateway_block):

"""

Args:

For backward compatibility, a sequence of numpy type names is also

accepted as an io signature.

"""

def __init__(self, name, in_sig, out_sig, interp):

gateway_block.__init__(self,

name=name,

self.gateway.set_relative_rate(self._interp, self._decim)

self.gateway.set_output_multiple(self._interp)

if (nitems > 0):

self.gateway.consume_each(int(nitems / self._interp))

@functools.wraps(func)

def wrapped(self, *points):

if not points:

elif len(points) == 1:

try:

block = points[0].to_basic_block()

except AttributeError:

func(self, block)

else:

try:

"""

Pass-through member requests to the C++ object.

"""

try:

object.__getattribute__(self, "_impl")

except AttributeError as exception:

"a derived class?".format(object.__getattribute__(self.__class__, "__name__"))) from exception

return getattr(self._impl, name)

# FIXME: these should really be implemented

# in the original C++ class (gr_hier_block2), then they would all be inherited here

from gnuradio import gr

import pmt

def make_lengthtags(lengths, offsets, tagname='length', vlen=1):

tags = []

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

tags.append(tag)

return tags

def string_to_vector(string):

v = []

for s in string:

v.append(ord(s))

return v

def strings_to_vectors(strings, tsb_tag_key):

vs = [string_to_vector(string) for string in strings]

return packets_to_vectors(vs, tsb_tag_key)

def vector_to_string(v):

s = []

for d in v:

s.append(chr(d))

return ''.join(s)

def vectors_to_strings(data, tags, tsb_tag_key):

packets = vectors_to_packets(data, tags, tsb_tag_key)

return [vector_to_string(packet) for packet in packets]

def count_bursts(data, tags, tsb_tag_key, vlen=1):

lengthtags = [t for t in tags

if pmt.symbol_to_string(t.key) == tsb_tag_key]

raise ValueError(

"More than one tags with key {0} with the same offset={1}."

.format(tsb_tag_key, tag.offset))

in_burst = False

in_packet = False

packet_length = None

for pos in range(len(data)):

if pos in lengths:

if in_packet:

raise Exception("Received packet tag while in packet.")

packet_pos = -1

packet_length = lengths[pos]

in_burst = False

if in_packet:

packet_pos += 1

in_packet = False

packet_pos = None

return burst_count

def vectors_to_packets(data, tags, tsb_tag_key, vlen=1):

lengthtags = [t for t in tags

if pmt.symbol_to_string(t.key) == tsb_tag_key]

raise ValueError(

"More than one tags with key {0} with the same offset={1}."

.format(tsb_tag_key, tag.offset))

if 0 not in lengths:

raise ValueError("There is no tag with key {0} and an offset of 0"

.format(tsb_tag_key))

length = lengths[pos]

if length == 0:

raise ValueError("Packets cannot have zero length.")

raise ValueError("The final packet is incomplete.")

pos += length

return packets

def packets_to_vectors(packets, tsb_tag_key, vlen=1):

""" Returns a single data vector and a set of tags.

If used with blocks.vector_source_X, this set of data

tags.append(tag)

offset = offset + len(packet)

return data, tags

class pubsub(dict):

def __init__(self):

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

dict.__setitem__(self, key, value)

sub(val)

def __getitem__(self, key):

if self._proxies[key] is not None:

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

return p[newkey]

return dict.__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):

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

def unproxy(self, key):

self._proxies[key] = None

# Test code

if __name__ == "__main__":

import sys

class subber(object):

def __init__(self, param):

self._param = param

def printer(self, x):

print(self._param, repr(x))

s = subber('param')

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

# The third is a lambda function

# Update key 'foo', will notify subscribers

print("Updating 'foo' with three subscribers:")

# Remove first subscriber

o.unsubscribe('foo', print_len)

# Update now will only trigger second and third subscriber

print("Updating 'foo' after removing a subscriber:")

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

print("Published value of 'baz':", o['baz'])

# Unpublish the key

self.tb.start()

self.tb.stop()

if __name__ == '__main__':

gr_unittest.run(test_flowgraph)

def test_010_end_with_head(self):

import math

exp = 1j * 440 / 44100

head = blocks.head(gr.sizeof_gr_complex, 1000)

test = test_hblk([gr.sizeof_gr_complex], 0)

tb = gr.top_block()

def test_011_test_message_connect(self):

import math

exp = 1j * 440 / 44100

strobe = blocks.message_strobe(pmt.PMT_NIL, 100)

head = blocks.head(gr.sizeof_gr_complex, 1000)

test = test_hblk([gr.sizeof_gr_complex], 1)

def test_012(self):

import math

exp = 1j * 440 / 44100

strobe = blocks.message_strobe(pmt.PMT_NIL, 100)

head = blocks.head(gr.sizeof_gr_complex, 1000)

test = test_hblk([gr.sizeof_gr_complex], 16)

N = 10**6

self.assertEqual(gr.xoroshiro128p_prng.min(), 0)

rng = gr.xoroshiro128p_prng(42)

arr = all((0 <= rng() <= 2**64 - 1 for _ in range(N)))

self.assertTrue(arr)

if __name__ == '__main__':

gr_unittest.run(test_random)

from . import gr_python as gr

class PythonTag(object):

" Python container for tags "

def __init__(self):

self.offset = None

def tag_to_python(tag):

""" Convert a stream tag to a Python-readable object """

newtag.srcid = pmt.to_python(tag.srcid)

return newtag

def python_to_tag(tag_struct):

"""

Convert a Python list/tuple/dictionary to a stream tag.

from .gr_python import (top_block_pb,

from .hier_block2 import hier_block2

import threading

from .hier_block2 import hier_block2

class _top_block_waiter(threading.Thread):

"""

This kludge allows ^C to interrupt top_block.run and top_block.wait

See also top_block.wait (below), which uses this class to implement

the interruptible wait.

"""

def __init__(self, tb):

threading.Thread.__init__(self)

self.setDaemon(1)

# We allow snakeCase here for consistency with unittest

# pylint: disable=invalid-name

class TestCase(unittest.TestCase):

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

Note that decimal places (from zero) is usually not the same

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

"""

raise self.failureException(

msg or '%r != %r within %r places' % (first, second, places))

raise self.failureException(

msg or '%r != %r within %r places' % (first, second, places)

)

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

"""

Fail if the two complex objects are unequal as determined by both

return

if abs(ref) > abs_eps:

raise self.failureException(

msg or '%r != %r rel_error = %r rel_limit = %r' % (

)

)

else:

raise self.failureException(

msg or '%r != %r rel_error = %r rel_limit = %r' % (

)

)

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

"""

Fail if the two complex tuples are not approximately equal.

for (x, y) in zip(a, b)

])

def assertComplexTuplesAlmostEqual2(self, a, b,

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

"""

for (x, y) in zip(a, b)

])

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

"""

Fail if the two real-valued tuples are not approximately equal.

for (x, y) in zip(a, b)

])

def assertFloatTuplesAlmostEqual2(self, a, b,

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

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

self.assertComplexAlmostEqual2(x, y, abs_eps, rel_eps, msg)

for (x, y) in zip(a, b)

])

def assertSequenceEqualGR(self, data_in, data_out):

Note this function exists because of this bug: https://bugs.python.org/issue19217

Calling self.assertEqual(seqA, seqB) can hang if seqA and seqB are not equal.

"""

if len(data_in) != len(data_out):

self.assertTrue(len(data_in) == len(data_out))

total_miscompares = 0

for idx, item in enumerate(zip(data_in, data_out)):

if item[0] != item[1]:

total_miscompares += 1

if total_miscompares > 0:

print('Total miscompares: {:d}'.format(total_miscompares))

self.assertTrue(total_miscompares == 0)

def waitFor(

self,

# getting the singleton locally.

PMT_NIL = pmt.get_PMT_NIL()

def pmt_to_tuple(p):

elems = list()

for i in range(pmt.length(p)):

elems.append(pmt_to_python(elem))

return tuple(elems)

def pmt_from_tuple(p):

args = list(map(python_to_pmt, p))

return pmt.make_tuple(*args)

def pmt_to_vector(p):

v = list()

for i in range(pmt.length(p)):

v.append(pmt_to_python(elem))

return v

def pmt_from_vector(p):

v = pmt.make_vector(len(p), PMT_NIL)

for i, elem in enumerate(p):

pmt.vector_set(v, i, python_to_pmt(elem))

return v

def pmt_to_dict(p):

d = dict()

items = pmt.dict_items(p)

d[pmt_to_python(k)] = pmt_to_python(v)

return d

def pmt_from_dict(p):

d = pmt.make_dict()

for k, v in list(p.items()):

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

return d

numpy_mappings = {

numpy.dtype(numpy.float32): (pmt.init_f32vector, float, pmt.f32vector_elements, pmt.is_f32vector),

numpy.dtype(numpy.float64): (pmt.init_f64vector, float, pmt.f64vector_elements, pmt.is_f64vector),

numpy.dtype(numpy.int8): (pmt.init_s8vector, int, pmt.s8vector_elements, pmt.is_s8vector),

numpy.dtype(numpy.int16): (pmt.init_s16vector, int, pmt.s16vector_elements, pmt.is_s16vector),

numpy.dtype(numpy.int32): (pmt.init_s32vector, int, pmt.s32vector_elements, pmt.is_s32vector),

numpy.dtype(numpy.uint8): (pmt.init_u8vector, int, pmt.u8vector_elements, pmt.is_u8vector),

numpy.dtype(numpy.uint16): (pmt.init_u16vector, int, pmt.u16vector_elements, pmt.is_u16vector),

numpy.dtype(numpy.uint32): (pmt.init_u32vector, int, pmt.u32vector_elements, pmt.is_u32vector),

numpy.dtype(numpy.byte): (pmt.init_u8vector, int, pmt.u8vector_elements, pmt.is_u8vector),

}

def numpy_to_uvector(numpy_array):

try:

pc = list(map(mapping[1], numpy.ravel(numpy_array)))

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

except KeyError:

def uvector_to_numpy(uvector):

(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),

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

(list, pmt.is_vector, pmt_to_vector, pmt_from_vector),

(dict, pmt.is_dict, pmt_to_dict, pmt_from_dict),

(numpy.ndarray, pmt.is_uniform_vector, uvector_to_numpy, numpy_to_uvector),

)

def pmt_to_python(p):

for python_type, pmt_check, to_python, from_python in type_mappings:

if pmt_check(p):

try:

return to_python(p)

# This exception will be handled by the general failure case

pass

def python_to_pmt(p):

for python_type, pmt_check, to_python, from_python in type_mappings:

if python_type is None: