summaryrefslogtreecommitdiff
path: root/gr-blocks/python/blocks/qa_ctrlport_probes.py
diff options
context:
space:
mode:
authorTom Rondeau <tom@trondeau.com>2014-08-08 11:28:05 -0400
committerTom Rondeau <tom@trondeau.com>2014-08-08 16:37:43 -0400
commit3d5df0ddd3aa8d5a94285b95f487747f25d4ee06 (patch)
tree715007cafdc7ac2b6eb954562be0b3a4a4612e2e /gr-blocks/python/blocks/qa_ctrlport_probes.py
parent3d18f70c66c974a82c5096acc4cbd37a47b6b55c (diff)
controlport: removing use of ice for a controlport rpc.
This effectively disables the use of ControlPort for now until we build in a new middleware layer. The ControlPort API and interfaces exist but will function as nops for now.
Diffstat (limited to 'gr-blocks/python/blocks/qa_ctrlport_probes.py')
-rw-r--r--gr-blocks/python/blocks/qa_ctrlport_probes.py187
1 files changed, 6 insertions, 181 deletions
diff --git a/gr-blocks/python/blocks/qa_ctrlport_probes.py b/gr-blocks/python/blocks/qa_ctrlport_probes.py
index b31934f705..91d96010fd 100644
--- a/gr-blocks/python/blocks/qa_ctrlport_probes.py
+++ b/gr-blocks/python/blocks/qa_ctrlport_probes.py
@@ -20,12 +20,9 @@
# Boston, MA 02110-1301, USA.
#
-import Ice
import sys, time, random, numpy
from gnuradio import gr, gr_unittest, blocks
-from gnuradio.ctrlport import GNURadio
-from gnuradio import ctrlport
import os, struct
class test_ctrlport_probes(gr_unittest.TestCase):
@@ -37,193 +34,21 @@ class test_ctrlport_probes(gr_unittest.TestCase):
def tearDown(self):
self.tb = None
- 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, 0, 2, 0, 3, 0, 4, 0,
- 5, 0, 6, 0, 7, 0, 8, 0]
-
- # Make sure we have time for flowgraph to run
- time.sleep(0.1)
-
- # Get available endpoint
- ep = gr.rpcmanager_get().endpoints()[0]
-
- # Initialize a simple Ice client from endpoint
- ic = Ice.initialize(sys.argv)
- base = ic.stringToProxy(ep)
- radio = GNURadio.ControlPortPrx.checkedCast(base)
-
- # Get all exported knobs
- ret = radio.get([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()
-
+ def xtest_001(self):
+ pass
def test_002(self):
- 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]
-
- # Initialize a simple Ice client from endpoint
- ic = Ice.initialize(sys.argv)
- base = ic.stringToProxy(ep)
- radio = GNURadio.ControlPortPrx.checkedCast(base)
-
- # Get all exported knobs
- ret = radio.get([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()
+ pass
def test_003(self):
- 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]
-
- # Initialize a simple Ice client from endpoint
- ic = Ice.initialize(sys.argv)
- base = ic.stringToProxy(ep)
- radio = GNURadio.ControlPortPrx.checkedCast(base)
-
- # Get all exported knobs
- ret = radio.get([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()
-
+ pass
def test_004(self):
- 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]
-
- # Initialize a simple Ice client from endpoint
- ic = Ice.initialize(sys.argv)
- base = ic.stringToProxy(ep)
- radio = GNURadio.ControlPortPrx.checkedCast(base)
-
- # Get all exported knobs
- ret = radio.get([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()
+ pass
def test_005(self):
- 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]
-
- # Initialize a simple Ice client from endpoint
- ic = Ice.initialize(sys.argv)
- base = ic.stringToProxy(ep)
- radio = GNURadio.ControlPortPrx.checkedCast(base)
-
- # Get all exported knobs
- ret = radio.get([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()
+ pass
if __name__ == '__main__':
gr_unittest.run(test_ctrlport_probes, "test_ctrlport_probes.xml")
-
generated by cgit v1.2.3 (git 2.39.1) at 2025-03-23 12:52:21 +0000