GR_PYTHON_INSTALL(FILES

__init__.py

tag_utils.py

packet_utils.py

gateway.py

top_block.py

pubsub.py

DESTINATION ${GR_PYTHON_DIR}/gnuradio/gr

)

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

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,

#

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

"""