git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/python/gnuradio/gr/qa_hier_block2.py

   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.main()