root / gnuradio-core / src / python / gnuradio / gr / qa_hier_block2.py @ 4c589268
History | View | Annotate | Download (13.3 kB)
| 1 | #!/usr/bin/env python
|
|---|---|
| 2 | |
| 3 | from gnuradio import gr, gr_unittest |
| 4 | |
| 5 | class test_hier_block2(gr_unittest.TestCase): |
| 6 | |
| 7 | def setUp(self): |
| 8 | pass
|
| 9 | |
| 10 | def tearDown(self): |
| 11 | pass
|
| 12 | |
| 13 | def test_001_make(self): |
| 14 | hblock = gr.hier_block2("test_block",
|
| 15 | gr.io_signature(1,1,gr.sizeof_int), |
| 16 | gr.io_signature(1,1,gr.sizeof_int)) |
| 17 | self.assertEqual("test_block", hblock.name()) |
| 18 | self.assertEqual(1, hblock.input_signature().max_streams()) |
| 19 | self.assertEqual(1, hblock.output_signature().min_streams()) |
| 20 | self.assertEqual(1, hblock.output_signature().max_streams()) |
| 21 | self.assertEqual(gr.sizeof_int, hblock.output_signature().sizeof_stream_item(0)) |
| 22 | |
| 23 | def test_002_connect_input(self): |
| 24 | hblock = gr.hier_block2("test_block",
|
| 25 | gr.io_signature(1,1,gr.sizeof_int), |
| 26 | gr.io_signature(1,1,gr.sizeof_int)) |
| 27 | nop1 = gr.nop(gr.sizeof_int) |
| 28 | hblock.connect(hblock, nop1) |
| 29 | |
| 30 | def test_004_connect_output(self): |
| 31 | hblock = gr.hier_block2("test_block",
|
| 32 | gr.io_signature(1,1,gr.sizeof_int), |
| 33 | gr.io_signature(1,1,gr.sizeof_int)) |
| 34 | nop1 = gr.nop(gr.sizeof_int) |
| 35 | hblock.connect(nop1, hblock) |
| 36 | |
| 37 | def test_005_connect_output_in_use(self): |
| 38 | hblock = gr.hier_block2("test_block",
|
| 39 | gr.io_signature(1,1,gr.sizeof_int), |
| 40 | gr.io_signature(1,1,gr.sizeof_int)) |
| 41 | nop1 = gr.nop(gr.sizeof_int) |
| 42 | nop2 = gr.nop(gr.sizeof_int) |
| 43 | hblock.connect(nop1, hblock) |
| 44 | self.assertRaises(ValueError, |
| 45 | lambda: hblock.connect(nop2, hblock))
|
| 46 | |
| 47 | def test_006_connect_invalid_src_port_neg(self): |
| 48 | hblock = gr.hier_block2("test_block",
|
| 49 | gr.io_signature(1,1,gr.sizeof_int), |
| 50 | gr.io_signature(1,1,gr.sizeof_int)) |
| 51 | nop1 = gr.nop(gr.sizeof_int) |
| 52 | self.assertRaises(ValueError, |
| 53 | lambda: hblock.connect((hblock, -1), nop1)) |
| 54 | |
| 55 | def test_005_connect_invalid_src_port_exceeds(self): |
| 56 | hblock = gr.hier_block2("test_block",
|
| 57 | gr.io_signature(1,1,gr.sizeof_int), |
| 58 | gr.io_signature(1,1,gr.sizeof_int)) |
| 59 | nop1 = gr.nop(gr.sizeof_int) |
| 60 | self.assertRaises(ValueError, |
| 61 | lambda: hblock.connect((hblock, 1), nop1)) |
| 62 | |
| 63 | def test_007_connect_invalid_dst_port_neg(self): |
| 64 | hblock = gr.hier_block2("test_block",
|
| 65 | gr.io_signature(1,1,gr.sizeof_int), |
| 66 | gr.io_signature(1,1,gr.sizeof_int)) |
| 67 | nop1 = gr.nop(gr.sizeof_int) |
| 68 | nop2 = gr.nop(gr.sizeof_int) |
| 69 | self.assertRaises(ValueError, |
| 70 | lambda: hblock.connect(nop1, (nop2, -1))) |
| 71 | |
| 72 | def test_008_connect_invalid_dst_port_exceeds(self): |
| 73 | hblock = gr.hier_block2("test_block",
|
| 74 | gr.io_signature(1,1,gr.sizeof_int), |
| 75 | gr.io_signature(1,1,gr.sizeof_int)) |
| 76 | nop1 = gr.null_sink(gr.sizeof_int) |
| 77 | nop2 = gr.null_sink(gr.sizeof_int) |
| 78 | self.assertRaises(ValueError, |
| 79 | lambda: hblock.connect(nop1, (nop2, 1))) |
| 80 | |
| 81 | def test_009_check_topology(self): |
| 82 | hblock = gr.top_block("test_block")
|
| 83 | hblock.check_topology(0, 0) |
| 84 | |
| 85 | def test_010_run(self): |
| 86 | expected = (1.0, 2.0, 3.0, 4.0) |
| 87 | hblock = gr.top_block("test_block")
|
| 88 | src = gr.vector_source_f(expected, False)
|
| 89 | sink1 = gr.vector_sink_f() |
| 90 | sink2 = gr.vector_sink_f() |
| 91 | hblock.connect(src, sink1) |
| 92 | hblock.connect(src, sink2) |
| 93 | hblock.run() |
| 94 | actual1 = sink1.data() |
| 95 | actual2 = sink2.data() |
| 96 | self.assertEquals(expected, actual1)
|
| 97 | self.assertEquals(expected, actual2)
|
| 98 | |
| 99 | def test_012_disconnect_input(self): |
| 100 | hblock = gr.hier_block2("test_block",
|
| 101 | gr.io_signature(1,1,gr.sizeof_int), |
| 102 | gr.io_signature(1,1,gr.sizeof_int)) |
| 103 | nop1 = gr.nop(gr.sizeof_int) |
| 104 | hblock.connect(hblock, nop1) |
| 105 | hblock.disconnect(hblock, nop1) |
| 106 | |
| 107 | def test_013_disconnect_input_not_connected(self): |
| 108 | hblock = gr.hier_block2("test_block",
|
| 109 | gr.io_signature(1,1,gr.sizeof_int), |
| 110 | gr.io_signature(1,1,gr.sizeof_int)) |
| 111 | nop1 = gr.nop(gr.sizeof_int) |
| 112 | nop2 = gr.nop(gr.sizeof_int) |
| 113 | hblock.connect(hblock, nop1) |
| 114 | self.assertRaises(ValueError, |
| 115 | lambda: hblock.disconnect(hblock, nop2))
|
| 116 | |
| 117 | def test_014_disconnect_input_neg(self): |
| 118 | hblock = gr.hier_block2("test_block",
|
| 119 | gr.io_signature(1,1,gr.sizeof_int), |
| 120 | gr.io_signature(1,1,gr.sizeof_int)) |
| 121 | nop1 = gr.nop(gr.sizeof_int) |
| 122 | hblock.connect(hblock, nop1) |
| 123 | self.assertRaises(ValueError, |
| 124 | lambda: hblock.disconnect((hblock, -1), nop1)) |
| 125 | |
| 126 | def test_015_disconnect_input_exceeds(self): |
| 127 | hblock = gr.hier_block2("test_block",
|
| 128 | gr.io_signature(1,1,gr.sizeof_int), |
| 129 | gr.io_signature(1,1,gr.sizeof_int)) |
| 130 | nop1 = gr.nop(gr.sizeof_int) |
| 131 | hblock.connect(hblock, nop1) |
| 132 | self.assertRaises(ValueError, |
| 133 | lambda: hblock.disconnect((hblock, 1), nop1)) |
| 134 | |
| 135 | def test_016_disconnect_output(self): |
| 136 | hblock = gr.hier_block2("test_block",
|
| 137 | gr.io_signature(1,1,gr.sizeof_int), |
| 138 | gr.io_signature(1,1,gr.sizeof_int)) |
| 139 | nop1 = gr.nop(gr.sizeof_int) |
| 140 | hblock.connect(nop1, hblock) |
| 141 | hblock.disconnect(nop1, hblock) |
| 142 | |
| 143 | def test_017_disconnect_output_not_connected(self): |
| 144 | hblock = gr.hier_block2("test_block",
|
| 145 | gr.io_signature(1,1,gr.sizeof_int), |
| 146 | gr.io_signature(1,1,gr.sizeof_int)) |
| 147 | nop1 = gr.nop(gr.sizeof_int) |
| 148 | nop2 = gr.nop(gr.sizeof_int) |
| 149 | hblock.connect(nop1, hblock) |
| 150 | self.assertRaises(ValueError, |
| 151 | lambda: hblock.disconnect(nop2, hblock))
|
| 152 | |
| 153 | def test_018_disconnect_output_neg(self): |
| 154 | hblock = gr.hier_block2("test_block",
|
| 155 | gr.io_signature(1,1,gr.sizeof_int), |
| 156 | gr.io_signature(1,1,gr.sizeof_int)) |
| 157 | nop1 = gr.nop(gr.sizeof_int) |
| 158 | hblock.connect(hblock, nop1) |
| 159 | self.assertRaises(ValueError, |
| 160 | lambda: hblock.disconnect(nop1, (hblock, -1))) |
| 161 | |
| 162 | def test_019_disconnect_output_exceeds(self): |
| 163 | hblock = gr.hier_block2("test_block",
|
| 164 | gr.io_signature(1,1,gr.sizeof_int), |
| 165 | gr.io_signature(1,1,gr.sizeof_int)) |
| 166 | nop1 = gr.nop(gr.sizeof_int) |
| 167 | hblock.connect(nop1, hblock) |
| 168 | self.assertRaises(ValueError, |
| 169 | lambda: hblock.disconnect(nop1, (hblock, 1))) |
| 170 | |
| 171 | def test_020_run(self): |
| 172 | hblock = gr.top_block("test_block")
|
| 173 | data = (1.0, 2.0, 3.0, 4.0) |
| 174 | src = gr.vector_source_f(data, False)
|
| 175 | dst = gr.vector_sink_f() |
| 176 | hblock.connect(src, dst) |
| 177 | hblock.run() |
| 178 | self.assertEquals(data, dst.data())
|
| 179 | |
| 180 | def test_021_connect_single(self): |
| 181 | hblock = gr.top_block("test_block")
|
| 182 | blk = gr.hier_block2("block",
|
| 183 | gr.io_signature(0, 0, 0), |
| 184 | gr.io_signature(0, 0, 0)) |
| 185 | hblock.connect(blk) |
| 186 | |
| 187 | def test_022_connect_single_with_ports(self): |
| 188 | hblock = gr.top_block("test_block")
|
| 189 | blk = gr.hier_block2("block",
|
| 190 | gr.io_signature(1, 1, 1), |
| 191 | gr.io_signature(1, 1, 1)) |
| 192 | self.assertRaises(ValueError, |
| 193 | lambda: hblock.connect(blk))
|
| 194 | |
| 195 | def test_023_connect_single_twice(self): |
| 196 | hblock = gr.top_block("test_block")
|
| 197 | blk = gr.hier_block2("block",
|
| 198 | gr.io_signature(0, 0, 0), |
| 199 | gr.io_signature(0, 0, 0)) |
| 200 | hblock.connect(blk) |
| 201 | self.assertRaises(ValueError, |
| 202 | lambda: hblock.connect(blk))
|
| 203 | |
| 204 | def test_024_disconnect_single(self): |
| 205 | hblock = gr.top_block("test_block")
|
| 206 | blk = gr.hier_block2("block",
|
| 207 | gr.io_signature(0, 0, 0), |
| 208 | gr.io_signature(0, 0, 0)) |
| 209 | hblock.connect(blk) |
| 210 | hblock.disconnect(blk) |
| 211 | |
| 212 | def test_025_disconnect_single_not_connected(self): |
| 213 | hblock = gr.top_block("test_block")
|
| 214 | blk = gr.hier_block2("block",
|
| 215 | gr.io_signature(0, 0, 0), |
| 216 | gr.io_signature(0, 0, 0)) |
| 217 | self.assertRaises(ValueError, |
| 218 | lambda: hblock.disconnect(blk))
|
| 219 | |
| 220 | def test_026_run_single(self): |
| 221 | expected_data = (1.0,)
|
| 222 | tb = gr.top_block("top_block")
|
| 223 | hb = gr.hier_block2("block",
|
| 224 | gr.io_signature(0, 0, 0), |
| 225 | gr.io_signature(0, 0, 0)) |
| 226 | src = gr.vector_source_f(expected_data) |
| 227 | dst = gr.vector_sink_f() |
| 228 | hb.connect(src, dst) |
| 229 | tb.connect(hb) |
| 230 | tb.run() |
| 231 | self.assertEquals(expected_data, dst.data())
|
| 232 | |
| 233 | def test_027a_internally_unconnected_input(self): |
| 234 | tb = gr.top_block() |
| 235 | hb = gr.hier_block2("block",
|
| 236 | gr.io_signature(1, 1, 1), |
| 237 | gr.io_signature(1, 1, 1)) |
| 238 | hsrc = gr.vector_source_b([1,])
|
| 239 | hb.connect(hsrc, hb) # wire output internally
|
| 240 | src = gr.vector_source_b([1, ])
|
| 241 | dst = gr.vector_sink_b() |
| 242 | tb.connect(src, hb, dst) # hb's input is not connected internally
|
| 243 | self.assertRaises(RuntimeError, |
| 244 | lambda: tb.run())
|
| 245 | |
| 246 | def test_027b_internally_unconnected_output(self): |
| 247 | tb = gr.top_block() |
| 248 | |
| 249 | hb = gr.hier_block2("block",
|
| 250 | gr.io_signature(1, 1, 1), |
| 251 | gr.io_signature(1, 1, 1)) |
| 252 | hdst = gr.vector_sink_b() |
| 253 | hb.connect(hb, hdst) # wire input internally
|
| 254 | src = gr.vector_source_b([1, ])
|
| 255 | dst = gr.vector_sink_b() |
| 256 | tb.connect(src, hb, dst) # hb's output is not connected internally
|
| 257 | self.assertRaises(RuntimeError, |
| 258 | lambda: tb.run())
|
| 259 | |
| 260 | def test_027c_fully_unconnected_output(self): |
| 261 | tb = gr.top_block() |
| 262 | hb = gr.hier_block2("block",
|
| 263 | gr.io_signature(1, 1, 1), |
| 264 | gr.io_signature(1, 1, 1)) |
| 265 | hsrc = gr.vector_sink_b() |
| 266 | hb.connect(hb, hsrc) # wire input internally
|
| 267 | src = gr.vector_source_b([1, ])
|
| 268 | dst = gr.vector_sink_b() |
| 269 | tb.connect(src, hb) # hb's output is not connected internally or externally
|
| 270 | self.assertRaises(RuntimeError, |
| 271 | lambda: tb.run())
|
| 272 | |
| 273 | def test_027d_fully_unconnected_input(self): |
| 274 | tb = gr.top_block() |
| 275 | hb = gr.hier_block2("block",
|
| 276 | gr.io_signature(1, 1, 1), |
| 277 | gr.io_signature(1, 1, 1)) |
| 278 | hdst = gr.vector_source_b([1,])
|
| 279 | hb.connect(hdst, hb) # wire output internally
|
| 280 | dst = gr.vector_sink_b() |
| 281 | tb.connect(hb, dst) # hb's input is not connected internally or externally
|
| 282 | self.assertRaises(RuntimeError, |
| 283 | lambda: tb.run())
|
| 284 | |
| 285 | def test_028_singleton_reconfigure(self): |
| 286 | tb = gr.top_block() |
| 287 | hb = gr.hier_block2("block",
|
| 288 | gr.io_signature(0, 0, 0), gr.io_signature(0, 0, 0)) |
| 289 | src = gr.vector_source_b([1, ])
|
| 290 | dst = gr.vector_sink_b() |
| 291 | hb.connect(src, dst) |
| 292 | tb.connect(hb) # Singleton connect
|
| 293 | tb.lock() |
| 294 | tb.disconnect_all() |
| 295 | tb.connect(src, dst) |
| 296 | tb.unlock() |
| 297 | |
| 298 | def test_029_singleton_disconnect(self): |
| 299 | tb = gr.top_block() |
| 300 | src = gr.vector_source_b([1, ])
|
| 301 | dst = gr.vector_sink_b() |
| 302 | tb.connect(src, dst) |
| 303 | tb.disconnect(src) # Singleton disconnect
|
| 304 | tb.connect(src, dst) |
| 305 | tb.run() |
| 306 | self.assertEquals(dst.data(), (1,)) |
| 307 | |
| 308 | def test_030_nested_input(self): |
| 309 | tb = gr.top_block() |
| 310 | src = gr.vector_source_b([1,])
|
| 311 | hb1 = gr.hier_block2("hb1",
|
| 312 | gr.io_signature(1, 1, gr.sizeof_char), |
| 313 | gr.io_signature(0, 0, 0)) |
| 314 | hb2 = gr.hier_block2("hb2",
|
| 315 | gr.io_signature(1, 1, gr.sizeof_char), |
| 316 | gr.io_signature(0, 0, 0)) |
| 317 | dst = gr.vector_sink_b() |
| 318 | tb.connect(src, hb1) |
| 319 | hb1.connect(hb1, hb2) |
| 320 | hb2.connect(hb2, gr.kludge_copy(gr.sizeof_char), dst) |
| 321 | tb.run() |
| 322 | self.assertEquals(dst.data(), (1,)) |
| 323 | |
| 324 | def test_031_multiple_internal_inputs(self): |
| 325 | tb = gr.top_block() |
| 326 | src = gr.vector_source_f([1.0,])
|
| 327 | hb = gr.hier_block2("hb",
|
| 328 | gr.io_signature(1, 1, gr.sizeof_float), |
| 329 | gr.io_signature(1, 1, gr.sizeof_float)) |
| 330 | m1 = gr.multiply_const_ff(1.0)
|
| 331 | m2 = gr.multiply_const_ff(2.0)
|
| 332 | add = gr.add_ff() |
| 333 | hb.connect(hb, m1) # m1 is connected to hb external input #0
|
| 334 | hb.connect(hb, m2) # m2 is also connected to hb external input #0
|
| 335 | hb.connect(m1, (add, 0))
|
| 336 | hb.connect(m2, (add, 1))
|
| 337 | hb.connect(add, hb) # add is connected to hb external output #0
|
| 338 | dst = gr.vector_sink_f() |
| 339 | tb.connect(src, hb, dst) |
| 340 | tb.run() |
| 341 | self.assertEquals(dst.data(), (3.0,)) |
| 342 | |
| 343 | def test_032_nested_multiple_internal_inputs(self): |
| 344 | tb = gr.top_block() |
| 345 | src = gr.vector_source_f([1.0,])
|
| 346 | hb = gr.hier_block2("hb",
|
| 347 | gr.io_signature(1, 1, gr.sizeof_float), |
| 348 | gr.io_signature(1, 1, gr.sizeof_float)) |
| 349 | hb2 = gr.hier_block2("hb",
|
| 350 | gr.io_signature(1, 1, gr.sizeof_float), |
| 351 | gr.io_signature(1, 1, gr.sizeof_float)) |
| 352 | |
| 353 | m1 = gr.multiply_const_ff(1.0)
|
| 354 | m2 = gr.multiply_const_ff(2.0)
|
| 355 | add = gr.add_ff() |
| 356 | hb2.connect(hb2, m1) # m1 is connected to hb2 external input #0
|
| 357 | hb2.connect(hb2, m2) # m2 is also connected to hb2 external input #0
|
| 358 | hb2.connect(m1, (add, 0))
|
| 359 | hb2.connect(m2, (add, 1))
|
| 360 | hb2.connect(add, hb2) # add is connected to hb2 external output #0
|
| 361 | hb.connect(hb, hb2, hb) # hb as hb2 as nested internal block
|
| 362 | dst = gr.vector_sink_f() |
| 363 | tb.connect(src, hb, dst) |
| 364 | tb.run() |
| 365 | self.assertEquals(dst.data(), (3.0,)) |
| 366 | |
| 367 | |
| 368 | if __name__ == "__main__": |
| 369 | gr_unittest.run(test_hier_block2, "test_hier_block2.xml")
|