Merge branch 'ctrlport'

author: Johnathan Corgan <johnathan@corganlabs.com> 2015-04-15 13:42:42 -0700
committer: Johnathan Corgan <johnathan@corganlabs.com> 2015-04-15 13:42:42 -0700
commit: b092142302bcf8c771ec68e61da7781eb406c86f (patch)
tree: 535b32403668d08323026939957187d30bda64a3 /gr-blocks/python
parent: 676d5cbe64f0b1293797025e5c4460f1e79896dc (diff)
parent: 0ab6607ffb0d45816a69c7069a1da90cc94bf047 (diff)
4 files changed, 528 insertions, 12 deletions
diff --git a/gr-blocks/python/blocks/CMakeLists.txt b/gr-blocks/python/blocks/CMakeLists.txt
index 44977313cf..19d808b1dd 100644
--- a/gr-blocks/python/blocks/CMakeLists.txt
+++ b/gr-blocks/python/blocks/CMakeLists.txt
@@ -43,14 +43,14 @@ if(ENABLE_TESTING)
   include(GrTest)
   file(GLOB py_qa_test_files "qa_*.py")
 
-  # Force out the controlport QA tests if we've disabled it.
-  if(NOT ENABLE_GR_CTRLPORT)
+  # Force out the controlport QA tests if we have no backends to use.
+  if(CTRLPORT_BACKENDS EQUAL 0)
     list(REMOVE_ITEM py_qa_test_files
       ${CMAKE_CURRENT_SOURCE_DIR}/qa_cpp_py_binding.py
       ${CMAKE_CURRENT_SOURCE_DIR}/qa_cpp_py_binding_set.py
       ${CMAKE_CURRENT_SOURCE_DIR}/qa_ctrlport_probes.py
       )
-  endif(NOT ENABLE_GR_CTRLPORT)
+  endif(CTRLPORT_BACKENDS EQUAL 0)
 
   foreach(py_qa_test_file ${py_qa_test_files})
     get_filename_component(py_qa_test_name ${py_qa_test_file} NAME_WE)
diff --git a/gr-blocks/python/blocks/qa_cpp_py_binding.py b/gr-blocks/python/blocks/qa_cpp_py_binding.py
new file mode 100644
index 0000000000..23a5c9b826
--- /dev/null
+++ b/gr-blocks/python/blocks/qa_cpp_py_binding.py
@@ -0,0 +1,173 @@
+#!/usr/bin/env python
+#
+# Copyright 2012,2013,2015 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# GNU Radio is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+#
+# This program tests mixed python and c++ ctrlport exports in a single app
+#
+
+import sys, time, random, numpy, re
+from gnuradio import gr, gr_unittest, blocks
+
+from gnuradio.ctrlport import GNURadio
+from gnuradio import ctrlport
+import os
+
+def get1():
+    return "success"
+
+def get2():
+    return "failure"
+
+class inc_class:
+    def __init__(self):
+        self.val = 1
+    def pp(self):
+        self.val = self.val+1
+        return self.val
+
+get3 = inc_class()
+
+def get4():
+    random.seed(0)
+    rv = random.random()
+    return rv
+
+def get5():
+    numpy.random.seed(0)
+    samp_t = numpy.random.randn(24)+1j*numpy.random.randn(24);
+    samp_f = numpy.fft.fft(samp_t);
+    log_pow_f = 20*numpy.log10(numpy.abs(samp_f))
+    rv = list(log_pow_f)
+    return rv;
+
+def get6():
+    numpy.random.seed(0)
+    samp_t = numpy.random.randn(1024)+1j*numpy.random.randn(1024);
+    rv = list(samp_t)
+    return rv;
+
+class test_cpp_py_binding(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gr.top_block()
+        os.environ['GR_CONF_CONTROLPORT_ON'] = 'True'
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_001(self):
+        v1 = gr.RPC_get_string("pyland", "v1", "unit_1_string",
+                               "Python Exported String", "", "", "",
+                               gr.DISPNULL)
+        v1.activate(get1)
+
+        v2 = gr.RPC_get_string("pyland", "v2", "unit_2_string",
+                               "Python Exported String", "", "", "",
+                               gr.DISPNULL)
+        v2.activate(get2)
+
+        v3 = gr.RPC_get_int("pyland", "v3", "unit_3_int",
+                            "Python Exported Int", 0, 100, 1,
+                            gr.DISPNULL)
+        v3.activate(get3.pp)
+
+        v4 = gr.RPC_get_double("pyland", "time", "unit_4_time_double",
+                               "Python Exported Double", 0, 1000, 1,
+                               gr.DISPNULL)
+        v4.activate(get4)
+
+        v5 = gr.RPC_get_vector_float("pyland", "fvec", "unit_5_float_vector",
+                                     "Python Exported Float Vector", [], [], [],
+                                     gr.DISPTIME | gr.DISPOPTCPLX)
+        v5.activate(get5)
+
+        v6 = gr.RPC_get_vector_gr_complex("pyland", "cvec", "unit_6_gr_complex_vector",
+                                          "Python Exported Complex Vector", [], [], [],
+                                          gr.DISPXY | gr.DISPOPTSCATTER)
+        v6.activate(get6)
+
+        # print some variables locally
+        val = get1()
+        rval = v1.get()
+        self.assertEqual(val, rval)
+
+        val = get2()
+        rval = v2.get()
+        self.assertEqual(val, rval)
+
+        val = get3.pp()
+        rval = v3.get()
+        self.assertEqual(val+1, rval)
+
+        val = get4()
+        rval = v4.get()
+        self.assertEqual(val, rval)
+
+        val = get5()
+        rval = v5.get()
+        self.assertComplexTuplesAlmostEqual(val, rval, 5)
+
+        val = get6()
+        rval = v6.get()
+        self.assertComplexTuplesAlmostEqual(val, rval, 5)
+
+    def test_002(self):
+        data = range(1,9)
+
+        self.src = blocks.vector_source_c(data)
+        self.p1 = blocks.ctrlport_probe_c("aaa","C++ exported variable")
+        self.p2 = blocks.ctrlport_probe_c("bbb","C++ exported variable")
+        probe_name = self.p2.alias()
+
+        self.tb.connect(self.src, self.p1)
+        self.tb.connect(self.src, self.p2)
+        self.tb.start()
+
+        # Probes return complex values as list of floats with re, im
+        # Imaginary parts of this data set are 0.
+        expected_result = [1, 2, 3, 4,
+                           5, 6, 7, 8]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::bbb"])
+        for name in ret.keys():
+            result = ret[name].value
+            self.assertEqual(result, expected_result)
+
+        self.tb.stop()
+
+if __name__ == '__main__':
+    gr_unittest.run(test_cpp_py_binding, "test_cpp_py_binding.xml")
diff --git a/gr-blocks/python/blocks/qa_cpp_py_binding_set.py b/gr-blocks/python/blocks/qa_cpp_py_binding_set.py
new file mode 100644
index 0000000000..5b81de08f9
--- /dev/null
+++ b/gr-blocks/python/blocks/qa_cpp_py_binding_set.py
@@ -0,0 +1,151 @@
+#!/usr/bin/env python
+#
+# Copyright 2012,2013,2015 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# GNU Radio is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+#
+# This program tests mixed python and c++ GRCP sets in a single app
+#
+
+import sys, time, random, numpy, re
+from gnuradio import gr, gr_unittest, blocks
+
+from gnuradio.ctrlport import GNURadio
+from gnuradio import ctrlport
+import os
+
+class inc_class:
+    def __init__(self,val):
+        self.val = val;
+
+    def _get(self):
+        #print "returning get (val = %s)"%(str(self.val));
+        return self.val;
+
+    def _set(self,val):
+        #print "updating val to %s"%(str(val));
+        self.val = val;
+        return;
+
+getset1 = inc_class(10);
+getset2 = inc_class(100.0);
+getset3 = inc_class("test");
+
+class test_cpp_py_binding_set(gr_unittest.TestCase):
+    def setUp(self):
+        self.tb = gr.top_block()
+        os.environ['GR_CONF_CONTROLPORT_ON'] = 'True'
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_001(self):
+
+        g1 = gr.RPC_get_int("pyland", "v1", "unit_1_int",
+                                  "Python Exported Int", 0, 100, 10,
+                                  gr.DISPNULL)
+        g1.activate(getset1._get)
+        s1 = gr.RPC_get_int("pyland", "v1", "unit_1_int",
+                                  "Python Exported Int", 0, 100, 10,
+                                  gr.DISPNULL)
+        s1.activate(getset1._set)
+        time.sleep(0.01)
+
+        # test int variables
+        getset1._set(21)
+        val = getset1._get()
+        rval = g1.get()
+        self.assertEqual(val, rval)
+
+        g2 = gr.RPC_get_float("pyland", "v2", "unit_2_float",
+                              "Python Exported Float", -100, 1000.0, 100.0,
+                              gr.DISPNULL)
+        g2.activate(getset2._get)
+        s2 = gr.RPC_get_float("pyland", "v2", "unit_2_float",
+                              "Python Exported Float", -100, 1000.0, 100.0,
+                              gr.DISPNULL)
+        s2.activate(getset2._set)
+        time.sleep(0.01)
+
+        # test float variables
+        getset2._set(123.456)
+        val = getset2._get()
+        rval = g2.get()
+        self.assertAlmostEqual(val, rval, 4)
+
+        g3 = gr.RPC_get_string("pyland", "v3", "unit_3_string",
+                               "Python Exported String", "", "", "",
+                               gr.DISPNULL)
+        g3.activate(getset3._get)
+        s3 = gr.RPC_get_string("pyland", "v3", "unit_3_string",
+                               "Python Exported String", "", "", "",
+                               gr.DISPNULL)
+        s3.activate(getset3._set)
+        time.sleep(0.01)
+
+        # test string variables
+        getset3._set("third test")
+        val = getset3._get()
+        rval = g3.get()
+        self.assertEqual(val, rval)
+
+
+    def test_002(self):
+        data = range(1, 10)
+
+        self.src = blocks.vector_source_c(data, True)
+        self.p = blocks.nop(gr.sizeof_gr_complex)
+        self.p.set_ctrlport_test(0);
+        probe_info = self.p.alias()
+
+        self.tb.connect(self.src, self.p)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        self.tb.start()
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get all exported knobs
+        key_name_test = probe_info+"::test"
+        ret = radio.getKnobs([key_name_test,])
+
+        ret[key_name_test].value = 10
+        radio.setKnobs({key_name_test: ret[key_name_test]})
+
+        ret = radio.getKnobs([])
+        result_test = ret[key_name_test].value
+        self.assertEqual(result_test, 10)
+
+        self.tb.stop()
+        self.tb.wait()
+
+if __name__ == '__main__':
+    gr_unittest.run(test_cpp_py_binding_set, "test_cpp_py_binding_set.xml")
diff --git a/gr-blocks/python/blocks/qa_ctrlport_probes.py b/gr-blocks/python/blocks/qa_ctrlport_probes.py
index 91d96010fd..c678846df0 100644
--- a/gr-blocks/python/blocks/qa_ctrlport_probes.py
+++ b/gr-blocks/python/blocks/qa_ctrlport_probes.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2013 Free Software Foundation, Inc.
+# Copyright 2013,2015 Free Software Foundation, Inc.
 #
 # This file is part of GNU Radio
 #
@@ -22,33 +22,225 @@
 
 import sys, time, random, numpy
 from gnuradio import gr, gr_unittest, blocks
+import os, struct, re
 
-import os, struct
+from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
 
 class test_ctrlport_probes(gr_unittest.TestCase):
 
     def setUp(self):
-        self.tb = gr.top_block()
         os.environ['GR_CONF_CONTROLPORT_ON'] = 'True'
+        self.tb = gr.top_block()
 
     def tearDown(self):
         self.tb = None
 
-    def xtest_001(self):
-        pass
+    def test_001(self):
+        data = range(1,9)
+
+        self.src = blocks.vector_source_c(data, True)
+        self.probe = blocks.ctrlport_probe2_c("samples","Complex",
+                                              len(data), gr.DISPNULL)
+        probe_name = self.probe.alias()
+
+        self.tb.connect(self.src, self.probe)
+        self.tb.start()
+
+
+        # Probes return complex values as list of floats with re, im
+        # Imaginary parts of this data set are 0.
+        expected_result = [1, 2, 3, 4,
+                           5, 6, 7, 8]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::samples"])
+        for name in ret.keys():
+            # Get data in probe, which might be offset; find the
+            # beginning and unwrap.
+            result = ret[name].value
+            i = result.index(complex(1.0, 0.0))
+            result = result[i:] + result[0:i]
+            self.assertComplexTuplesAlmostEqual(expected_result, result, 4)
+
+        self.tb.stop()
+        self.tb.wait()
+
 
     def test_002(self):
-        pass
+        data = range(1,9)
+
+        self.src = blocks.vector_source_f(data, True)
+        self.probe = blocks.ctrlport_probe2_f("samples","Floats",
+                                              len(data), gr.DISPNULL)
+        probe_name = self.probe.alias()
+
+        self.tb.connect(self.src, self.probe)
+        self.tb.start()
+
+        expected_result = [1, 2, 3, 4, 5, 6, 7, 8,]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
 
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::samples"])
+        for name in ret.keys():
+            # Get data in probe, which might be offset; find the
+            # beginning and unwrap.
+            result = ret[name].value
+            i = result.index(1.0)
+            result = result[i:] + result[0:i]
+            self.assertEqual(expected_result, result)
+
+        self.tb.stop()
+        self.tb.wait()
 
     def test_003(self):
-        pass
+        data = range(1,9)
+
+        self.src = blocks.vector_source_i(data, True)
+        self.probe = blocks.ctrlport_probe2_i("samples","Integers",
+                                              len(data), gr.DISPNULL)
+        probe_name = self.probe.alias()
+
+        self.tb.connect(self.src, self.probe)
+        self.tb.start()
+
+        expected_result = [1, 2, 3, 4, 5, 6, 7, 8,]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::samples"])
+        for name in ret.keys():
+            # Get data in probe, which might be offset; find the
+            # beginning and unwrap.
+            result = ret[name].value
+            i = result.index(1.0)
+            result = result[i:] + result[0:i]
+            self.assertEqual(expected_result, result)
+
+        self.tb.stop()
+        self.tb.wait()
+
 
     def test_004(self):
-        pass
+        data = range(1,9)
+
+        self.src = blocks.vector_source_s(data, True)
+        self.probe = blocks.ctrlport_probe2_s("samples","Shorts",
+                                              len(data), gr.DISPNULL)
+        probe_name = self.probe.alias()
+
+        self.tb.connect(self.src, self.probe)
+        self.tb.start()
+
+        expected_result = [1, 2, 3, 4, 5, 6, 7, 8,]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::samples"])
+        for name in ret.keys():
+            # Get data in probe, which might be offset; find the
+            # beginning and unwrap.
+            result = ret[name].value
+            i = result.index(1.0)
+            result = result[i:] + result[0:i]
+            self.assertEqual(expected_result, result)
+
+        self.tb.stop()
+        self.tb.wait()
 
     def test_005(self):
-        pass
+        data = range(1,9)
+
+        self.src = blocks.vector_source_b(data, True)
+        self.probe = blocks.ctrlport_probe2_b("samples","Bytes",
+                                              len(data), gr.DISPNULL)
+        probe_name = self.probe.alias()
+
+        self.tb.connect(self.src, self.probe)
+        self.tb.start()
+
+        expected_result = [1, 2, 3, 4, 5, 6, 7, 8,]
+
+        # Make sure we have time for flowgraph to run
+        time.sleep(0.1)
+
+        # Get available endpoint
+        ep = gr.rpcmanager_get().endpoints()[0]
+        hostname = re.search("-h (\S+|\d+\.\d+\.\d+\.\d+)", ep).group(1)
+        portnum = re.search("-p (\d+)", ep).group(1)
+        argv = [None, hostname, portnum]
+
+        # Initialize a simple ControlPort client from endpoint
+        from gnuradio.ctrlport.GNURadioControlPortClient import GNURadioControlPortClient
+        radiosys = GNURadioControlPortClient(argv=argv, rpcmethod='thrift')
+        radio = radiosys.client
+
+        # Get all exported knobs
+        ret = radio.getKnobs([probe_name + "::samples"])
+        for name in ret.keys():
+            # Get data in probe, which might be offset; find the
+            # beginning and unwrap.
+            result = ret[name].value
+            result = list(struct.unpack(len(result)*'b', result))
+            i = result.index(1)
+            result = result[i:] + result[0:i]
+            self.assertEqual(expected_result, result)
+
+        self.tb.stop()
+        self.tb.wait()
 
 if __name__ == '__main__':
     gr_unittest.run(test_ctrlport_probes, "test_ctrlport_probes.xml")
author	Johnathan Corgan <johnathan@corganlabs.com>	2015-04-15 13:42:42 -0700
committer	Johnathan Corgan <johnathan@corganlabs.com>	2015-04-15 13:42:42 -0700
commit	b092142302bcf8c771ec68e61da7781eb406c86f (patch)
tree	535b32403668d08323026939957187d30bda64a3 /gr-blocks/python
parent	676d5cbe64f0b1293797025e5c4460f1e79896dc (diff)
parent	0ab6607ffb0d45816a69c7069a1da90cc94bf047 (diff)