Merge branch 'master' into file_meta_data

Conflicts:
	gr-utils/src/python/CMakeLists.txt

7 files changed, 592 insertions, 1 deletions

diff --git a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
index 6a0555021..62f3d7e46 100644
--- a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
+++ b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
@@ -23,6 +23,7 @@ include(GrPython)
 GR_PYTHON_INSTALL(FILES
     __init__.py
     exceptions.py
+    gateway.py
     gr_threading.py
     gr_threading_23.py
     gr_threading_24.py
@@ -42,6 +43,13 @@ 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_SOURCE_DIR}/gruel/src/python
+        ${CMAKE_BINARY_DIR}/gruel/src/swig
+        ${CMAKE_BINARY_DIR}/gnuradio-core/src/python
+        ${CMAKE_BINARY_DIR}/gnuradio-core/src/lib/swig
+    )
+    set(GR_TEST_TARGET_DEPS volk gruel gnuradio-core)
     GR_ADD_TEST(${py_qa_test_name} ${PYTHON_EXECUTABLE} ${PYTHON_DASH_B} ${py_qa_test_file})
 endforeach(py_qa_test_file)
 endif(ENABLE_TESTING)
diff --git a/gnuradio-core/src/python/gnuradio/gr/__init__.py b/gnuradio-core/src/python/gnuradio/gr/__init__.py
index e5a8fdbf9..5d01ea11b 100644
--- a/gnuradio-core/src/python/gnuradio/gr/__init__.py
+++ b/gnuradio-core/src/python/gnuradio/gr/__init__.py
@@ -1,5 +1,5 @@
 #
-# Copyright 2003,2004,2006,2008,2009,2010 Free Software Foundation, Inc.
+# Copyright 2003-2012 Free Software Foundation, Inc.
 #
 # This file is part of GNU Radio
 #
@@ -28,6 +28,7 @@ from gnuradio_core import *
 from exceptions import *
 from hier_block2 import *
 from top_block import *
+from gateway import basic_block, sync_block, decim_block, interp_block
 
 # create a couple of aliases
 serial_to_parallel = stream_to_vector
diff --git a/gnuradio-core/src/python/gnuradio/gr/gateway.py b/gnuradio-core/src/python/gnuradio/gr/gateway.py
new file mode 100644
index 000000000..244b8b592
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/gateway.py
@@ -0,0 +1,215 @@
+#
+# 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.
+#
+
+import gnuradio_core as gr_core
+from gnuradio_core import io_signature, io_signaturev
+from gnuradio_core import gr_block_gw_message_type
+from gnuradio_core import block_gateway
+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):
+    class array_like:
+        __array_interface__ = {
+            'data' : (int(addr), False),
+            'typestr' : dtype.base.str,
+            'descr' : dtype.base.descr,
+            'shape' : (nitems,) + dtype.shape,
+            'strides' : None,
+            'version' : 3
+        }
+    return numpy.asarray(array_like()).view(dtype.base)
+
+########################################################################
+# Handler that does callbacks from C++
+########################################################################
+class gateway_handler(gr_core.feval_ll):
+
+    #dont put a constructor, it wont work
+
+    def init(self, callback):
+        self._callback = callback
+
+    def eval(self, arg):
+        try: self._callback()
+        except Exception as ex:
+            print("handler caught exception: %s"%ex)
+            import traceback; traceback.print_exc()
+            raise ex
+        return 0
+
+########################################################################
+# The guts that make this into a gr block
+########################################################################
+class gateway_block(object):
+
+    def __init__(self, name, in_sig, out_sig, work_type, factor):
+
+        #ensure that the sigs are iterable dtypes
+        def sig_to_dtype_sig(sig):
+            if sig is None: sig = ()
+            return map(numpy.dtype, 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
+        self.__in_indexes = range(len(self.__in_sig))
+        self.__out_indexes = range(len(self.__out_sig))
+
+        #convert the signatures into gr.io_signatures
+        def sig_to_gr_io_sigv(sig):
+            if not len(sig): return io_signature(0, 0, 0)
+            return io_signaturev(len(sig), len(sig), [s.itemsize for s in sig])
+        gr_in_sig = sig_to_gr_io_sigv(self.__in_sig)
+        gr_out_sig = sig_to_gr_io_sigv(self.__out_sig)
+
+        #create internal gateway block
+        self.__handler = gateway_handler()
+        self.__handler.init(self.__gr_block_handle)
+        self.__gateway = block_gateway(
+            self.__handler, name, gr_in_sig, gr_out_sig, work_type, factor)
+        self.__message = self.__gateway.gr_block_message()
+
+        #register gr_block functions
+        prefix = 'gr_block__'
+        for attr in [x for x in dir(self.__gateway) if x.startswith(prefix)]:
+            setattr(self, attr.replace(prefix, ''), getattr(self.__gateway, attr))
+        self.pop_msg_queue = lambda: gr_core.gr_block_gw_pop_msg_queue_safe(self.__gateway)
+
+    def to_basic_block(self):
+        """
+        Makes this block connectable by hier/top block python
+        """
+        return self.__gateway.to_basic_block()
+
+    def __gr_block_handle(self):
+        """
+        Dispatch tasks according to the action type specified in the message.
+        """
+        if self.__message.action == gr_block_gw_message_type.ACTION_GENERAL_WORK:
+            self.__message.general_work_args_return_value = self.general_work(
+
+                input_items=[pointer_to_ndarray(
+                    self.__message.general_work_args_input_items[i],
+                    self.__in_sig[i],
+                    self.__message.general_work_args_ninput_items[i]
+                ) for i in self.__in_indexes],
+
+                output_items=[pointer_to_ndarray(
+                    self.__message.general_work_args_output_items[i],
+                    self.__out_sig[i],
+                    self.__message.general_work_args_noutput_items
+                ) for i in self.__out_indexes],
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_WORK:
+            self.__message.work_args_return_value = self.work(
+
+                input_items=[pointer_to_ndarray(
+                    self.__message.work_args_input_items[i],
+                    self.__in_sig[i],
+                    self.__message.work_args_ninput_items
+                ) for i in self.__in_indexes],
+
+                output_items=[pointer_to_ndarray(
+                    self.__message.work_args_output_items[i],
+                    self.__out_sig[i],
+                    self.__message.work_args_noutput_items
+                ) for i in self.__out_indexes],
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_FORECAST:
+            self.forecast(
+                noutput_items=self.__message.forecast_args_noutput_items,
+                ninput_items_required=self.__message.forecast_args_ninput_items_required,
+            )
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_START:
+            self.__message.start_args_return_value = self.start()
+
+        elif self.__message.action == gr_block_gw_message_type.ACTION_STOP:
+            self.__message.stop_args_return_value = self.stop()
+
+    def forecast(self, noutput_items, ninput_items_required):
+        """
+        forecast is only called from a general block
+        this is the default implementation
+        """
+        for ninput_item in ninput_items_required:
+            ninput_item = noutput_items + self.history() - 1;
+        return
+
+    def general_work(self, *args, **kwargs):
+        """general work to be overloaded in a derived class"""
+        raise NotImplementedError("general work not implemented")
+
+    def work(self, *args, **kwargs):
+        """work to be overloaded in a derived class"""
+        raise NotImplementedError("work not implemented")
+
+    def start(self): return True
+    def stop(self): return True
+
+########################################################################
+# Wrappers for the user to inherit from
+########################################################################
+class basic_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_GENERAL,
+            factor=1, #not relevant factor
+        )
+
+class sync_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_SYNC,
+            factor=1,
+        )
+
+class decim_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig, decim):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_DECIM,
+            factor=decim,
+        )
+
+class interp_block(gateway_block):
+    def __init__(self, name, in_sig, out_sig, interp):
+        gateway_block.__init__(self,
+            name=name,
+            in_sig=in_sig,
+            out_sig=out_sig,
+            work_type=gr_core.GR_BLOCK_GW_WORK_INTERP,
+            factor=interp,
+        )
diff --git a/gnuradio-core/src/python/gnuradio/gr/hier_block2.py b/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
index 0c45f1691..b95782238 100644
--- a/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
+++ b/gnuradio-core/src/python/gnuradio/gr/hier_block2.py
@@ -20,6 +20,10 @@
 #
 
 from gnuradio_core import hier_block2_swig
+try:
+    import pmt
+except ImportError:
+    from gruel import pmt
 
 #
 # This hack forces a 'has-a' relationship to look like an 'is-a' one.
@@ -111,3 +115,15 @@ class hier_block2(object):
         self._hb.primitive_disconnect(src_block.to_basic_block(), src_port,
                                       dst_block.to_basic_block(), dst_port)
 
+    def msg_connect(self, src, srcport, dst, dstport):
+        self.primitive_msg_connect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
+    def msg_disconnect(self, src, srcport, dst, dstport):
+        self.primitive_msg_disconnect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
+    def message_port_register_hier_in(self, portname):
+        self.primitive_message_port_register_hier_in(pmt.pmt_intern(portname));
+
+    def message_port_register_hier_out(self, portname):
+        self.primitive_message_port_register_hier_out(pmt.pmt_intern(portname));
+
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py b/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py
new file mode 100644
index 000000000..911879f6f
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_block_gateway.py
@@ -0,0 +1,235 @@
+#
+# 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 pmt
+import numpy
+
+class add_2_f32_1_f32(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "add 2 f32",
+            in_sig = [numpy.float32, numpy.float32],
+            out_sig = [numpy.float32],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0] + input_items[1]
+        return len(output_items[0])
+
+class add_2_fc32_1_fc32(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "add 2 fc32",
+            in_sig = [numpy.complex64, numpy.complex64],
+            out_sig = [numpy.complex64],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0] + input_items[1]
+        return len(output_items[0])
+
+class convolve(gr.sync_block):
+    """
+    A demonstration using block history to properly perform a convolution.
+    """
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "convolve",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32]
+        )
+        self._taps = [1, 0, 0, 0]
+        self.set_history(len(self._taps))
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = numpy.convolve(input_items[0], self._taps, mode='valid')
+        return len(output_items[0])
+
+class decim2x(gr.decim_block):
+    def __init__(self):
+        gr.decim_block.__init__(
+            self,
+            name = "decim2x",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32],
+            decim = 2
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][:] = input_items[0][::2]
+        return len(output_items[0])
+
+class interp2x(gr.interp_block):
+    def __init__(self):
+        gr.interp_block.__init__(
+            self,
+            name = "interp2x",
+            in_sig = [numpy.float32],
+            out_sig = [numpy.float32],
+            interp = 2
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][1::2] = input_items[0]
+        output_items[0][::2] = input_items[0]
+        return len(output_items[0])
+
+class tag_source(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "tag source",
+            in_sig = None,
+            out_sig = [numpy.float32],
+        )
+
+    def work(self, input_items, output_items):
+        num_output_items = len(output_items[0])
+
+        #put code here to fill the output items...
+
+        #make a new tag on the middle element every time work is called
+        count = self.nitems_written(0) + num_output_items/2
+        key = pmt.pmt_string_to_symbol("example_key")
+        value = pmt.pmt_string_to_symbol("example_value")
+        self.add_item_tag(0, count, key, value)
+
+        return num_output_items
+
+class tag_sink(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "tag sink",
+            in_sig = [numpy.float32],
+            out_sig = None,
+        )
+        self.key = None
+
+    def work(self, input_items, output_items):
+        num_input_items = len(input_items[0])
+
+        #put code here to process the input items...
+
+        #print all the tags received in this work call
+        nread = self.nitems_read(0)
+        tags = self.get_tags_in_range(0, nread, nread+num_input_items)
+        for tag in tags:
+            #print tag.offset
+            #print pmt.pmt_symbol_to_string(tag.key)
+            #print pmt.pmt_symbol_to_string(tag.value)
+            self.key = pmt.pmt_symbol_to_string(tag.key)
+
+        return num_input_items
+
+class fc32_to_f32_2(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "fc32_to_f32_2",
+            in_sig = [numpy.complex64],
+            out_sig = [(numpy.float32, 2)],
+        )
+
+    def work(self, input_items, output_items):
+        output_items[0][::,0] = numpy.real(input_items[0])
+        output_items[0][::,1] = numpy.imag(input_items[0])
+        return len(output_items[0])
+
+class test_block_gateway(gr_unittest.TestCase):
+
+    def test_add_f32(self):
+        tb = gr.top_block()
+        src0 = gr.vector_source_f([1, 3, 5, 7, 9], False)
+        src1 = gr.vector_source_f([0, 2, 4, 6, 8], False)
+        adder = add_2_f32_1_f32()
+        sink = gr.vector_sink_f()
+        tb.connect((src0, 0), (adder, 0))
+        tb.connect((src1, 0), (adder, 1))
+        tb.connect(adder, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 5, 9, 13, 17))
+
+    def test_add_fc32(self):
+        tb = gr.top_block()
+        src0 = gr.vector_source_c([1, 3j, 5, 7j, 9], False)
+        src1 = gr.vector_source_c([0, 2j, 4, 6j, 8], False)
+        adder = add_2_fc32_1_fc32()
+        sink = gr.vector_sink_c()
+        tb.connect((src0, 0), (adder, 0))
+        tb.connect((src1, 0), (adder, 1))
+        tb.connect(adder, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 5j, 9, 13j, 17))
+
+    def test_convolve(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False)
+        cv = convolve()
+        sink = gr.vector_sink_f()
+        tb.connect(src, cv, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8))
+
+    def test_decim2x(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 2, 3, 4, 5, 6, 7, 8], False)
+        d2x = decim2x()
+        sink = gr.vector_sink_f()
+        tb.connect(src, d2x, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 3, 5, 7))
+
+    def test_interp2x(self):
+        tb = gr.top_block()
+        src = gr.vector_source_f([1, 3, 5, 7, 9], False)
+        i2x = interp2x()
+        sink = gr.vector_sink_f()
+        tb.connect(src, i2x, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 1, 3, 3, 5, 5, 7, 7, 9, 9))
+
+    def test_tags(self):
+        src = tag_source()
+        sink = tag_sink()
+        head = gr.head(gr.sizeof_float, 50000) #should be enough items to get a tag through
+        tb = gr.top_block()
+        tb.connect(src, head, sink)
+        tb.run()
+        self.assertEqual(sink.key, "example_key")
+
+    def test_fc32_to_f32_2(self):
+        tb = gr.top_block()
+        src = gr.vector_source_c([1+2j, 3+4j, 5+6j, 7+8j, 9+10j], False)
+        convert = fc32_to_f32_2()
+        v2s = gr.vector_to_stream(gr.sizeof_float, 2)
+        sink = gr.vector_sink_f()
+        tb.connect(src, convert, v2s, sink)
+        tb.run()
+        self.assertEqual(sink.data(), (1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+
+if __name__ == '__main__':
+    gr_unittest.run(test_block_gateway, "test_block_gateway.xml")
+
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
new file mode 100755
index 000000000..ebc365b61
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
@@ -0,0 +1,110 @@
+#!/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 pmt
+import time
+
+class test_pdu(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gr.top_block()
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_000(self):
+        # Just run some data through and make sure it doesn't puke.
+        src_data = (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+
+        src = gr.pdu_to_tagged_stream(gr.BYTE)
+        snk3 = gr.tagged_stream_to_pdu(gr.BYTE)
+        snk2 = gr.vector_sink_b()
+        snk = gr.tag_debug(1, "test")
+
+        dbg = gr.message_debug()
+
+        # Test that the right number of ports exist.
+        pi = dbg.message_ports_in()
+        po = dbg.message_ports_out()
+        self.assertEqual(pmt.pmt_length(pi), 2)
+        self.assertEqual(pmt.pmt_length(po), 0)
+
+        pi = snk3.message_ports_in()
+        po = snk3.message_ports_out()
+        self.assertEqual(pmt.pmt_length(pi), 0)
+        self.assertEqual(pmt.pmt_length(po), 1)
+
+        #print "Message Debug input ports: "
+        #pmt.pmt_print(pi)
+        #print "Message Debug output ports: "
+        #pmt.pmt_print(po)
+        
+        #print "Stream to PDU input ports: "
+        #pmt.pmt_print(pi)
+        #print "Stream to PDU output ports: "
+        #pmt.pmt_print(po)
+        time.sleep(0.1)
+
+        self.tb.connect(src, snk)
+        self.tb.connect(src, snk2)
+        self.tb.connect(src, snk3)
+
+        self.tb.msg_connect(snk3, "pdus", dbg, "store")
+        self.tb.start()
+
+        # make our reference and message pmts
+        port = pmt.pmt_intern("pdus")
+        msg = pmt.pmt_cons( pmt.PMT_NIL, pmt.pmt_make_u8vector(16, 0xFF) )
+        
+        #print "printing port & msg"
+        #pmt.pmt_print(port)
+        #pmt.pmt_print(msg)
+
+        # post the message
+        src.to_basic_block()._post( port, msg )
+
+        while(dbg.num_messages() < 1):
+            time.sleep(0.5)
+        self.tb.stop()
+        self.tb.wait()
+
+        # Get the vector of data from the vector sink
+        result_data = snk2.data()
+
+        # Get the vector of data from the message sink
+        # Convert the message PMT as a pair into its vector
+        result_msg = dbg.get_message(0)
+        msg_vec = pmt.pmt_cdr(result_msg)
+        pmt.pmt_print(msg_vec)
+
+        # Convert the PMT vector into a Python list
+        msg_data = []
+        for i in xrange(16):
+            msg_data.append(pmt.pmt_u8vector_ref(msg_vec, i))
+
+        actual_data = 16*[0xFF,]
+        self.assertEqual(actual_data, list(result_data))
+        self.assertEqual(actual_data, msg_data)
+
+if __name__ == '__main__':
+    gr_unittest.run(test_pdu, "test_pdu.xml")
diff --git a/gnuradio-core/src/python/gnuradio/gr/top_block.py b/gnuradio-core/src/python/gnuradio/gr/top_block.py
index 43af8073b..dc1f443aa 100644
--- a/gnuradio-core/src/python/gnuradio/gr/top_block.py
+++ b/gnuradio-core/src/python/gnuradio/gr/top_block.py
@@ -123,6 +123,12 @@ class top_block(object):
 		for i in range (1, len (points)):
         	    self._connect(points[i-1], points[i])
 
+    def msg_connect(self, src, srcport, dst, dstport):
+        self.primitive_msg_connect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+    
+    def msg_disconnect(self, src, srcport, dst, dstport):
+        self.primitive_msg_disconnect(src.to_basic_block(), srcport, dst.to_basic_block(), dstport);
+
     def _connect(self, src, dst):
         (src_block, src_port) = self._coerce_endpoint(src)
         (dst_block, dst_port) = self._coerce_endpoint(dst)