1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
# Copyright (C) by Josh Blum. See LICENSE.txt for licensing information.
import unittest
import gras
import numpy
from gras import TestUtils
class MyBlock(gras.Block):
def __init__(self):
gras.Block.__init__(self, "MyBlock", out_sig=[numpy.uint32], in_sig=[numpy.uint32])
self.numeric_value = 0
self.register_call("get_numeric_value", self.get_numeric_value)
self.register_call("set_numeric_value", self.set_numeric_value)
self.vector_value = [0]
self.register_call("get_vector_value", self.get_vector_value)
self.register_call("set_vector_value", self.set_vector_value)
def work(self, ins, outs):
n = min(len(ins[0]), len(outs[0]))
outs[0][:n] = ins[0][:n] + self.numeric_value
self.consume(n)
self.produce(n)
def get_numeric_value(self):
return self.numeric_value
def set_numeric_value(self, new_numeric_value):
print "new_numeric_value", new_numeric_value
self.numeric_value = new_numeric_value
def get_vector_value(self):
return self.vector_value
def set_vector_value(self, new_vector_value):
print "new_vector_value", new_vector_value
self.vector_value = numpy.copy(new_vector_value)
class QueryTest(unittest.TestCase):
def setUp(self):
self.tb = gras.TopBlock()
def tearDown(self):
self.tb = None
def test_simple(self):
vec_source = TestUtils.VectorSource(numpy.uint32, [0, 9, 8, 7, 6])
vec_sink = TestUtils.VectorSink(numpy.uint32)
self.tb.connect(vec_source, vec_sink)
self.tb.run()
self.assertEqual(vec_sink.data(), (0, 9, 8, 7, 6))
#query the block list
blocks_result = self.tb.query(dict(path="/blocks.json"))
self.assertEqual(len(blocks_result['blocks']), 2)
#pick a block to query below:
block_id = blocks_result['blocks'].keys()[0]
#query the stats
stats_result = self.tb.query(dict(
path="/stats.json",
blocks=[block_id],
))
self.assertTrue('tps' in stats_result)
self.assertTrue('now' in stats_result)
#found the block we asked for
self.assertTrue(block_id in stats_result['blocks'])
def test_numeric_query(self):
vec_source = TestUtils.VectorSource(numpy.uint32, [0, 9, 8, 7, 6])
vec_sink = TestUtils.VectorSink(numpy.uint32)
block = MyBlock()
block.set_uid("test_numeric_query")
self.tb.connect(vec_source, block, vec_sink)
self.tb.run()
#query the block list
blocks_result = self.tb.query(dict(path="/blocks.json"))
self.assertEqual(len(blocks_result['blocks']), 3)
self.assertTrue('test_numeric_query' in blocks_result['blocks'])
#set the integer property
self.tb.query(dict(
path="/calls.json",
block='test_numeric_query',
name='set_numeric_value',
args=[42],
))
self.assertEqual(block.numeric_value, 42)
#get the integer property
block.set_numeric_value(21)
result = self.tb.query(dict(
path="/calls.json",
block='test_numeric_query',
name='get_numeric_value',
))
self.assertEqual(result['value'], 21)
#set the complex property
self.tb.query(dict(
path="/calls.json",
block='test_numeric_query',
name='set_numeric_value',
args=['(0, 42)'],
))
self.assertEqual(block.numeric_value, 42j)
#get the complex property
block.set_numeric_value(21j)
result = self.tb.query(dict(
path="/calls.json",
block='test_numeric_query',
name='get_numeric_value',
))
self.assertEqual(result['value'], '(0,21)')
def test_vector_query(self):
vec_source = TestUtils.VectorSource(numpy.uint32, [0, 9, 8, 7, 6])
vec_sink = TestUtils.VectorSink(numpy.uint32)
block = MyBlock()
block.set_uid("test_vector_query")
self.tb.connect(vec_source, block, vec_sink)
self.tb.run()
#set the vector property
self.tb.query(dict(
path="/calls.json",
block='test_vector_query',
name='set_vector_value',
args=[[1, 2, 3, 4, 5]],
))
self.assertEqual(list(block.vector_value), [1, 2, 3, 4, 5])
#get the vector property
block.set_vector_value([6, 7, 8, 9])
result = self.tb.query(dict(
path="/calls.json",
block='test_vector_query',
name='get_vector_value',
))
self.assertEqual(list(result['value']), [6, 7, 8, 9])
if __name__ == '__main__':
unittest.main()
|
