-gr_blocktool: Block Header parsing tool.

-Parses GNU Radio header files to generate YAML or JSON output.

-

-This directory consists of test for the parsed header output and Blocktool Exceptions.

-============================================================================================

-- Schema defined in the test will be strictly followed for every parsed JSON output file.

-

-Two sample response for header files are available in this directory:

-=====================================================================

-

-- sample_agc2_cc.json for public header file agc2_cc.h from gr-analog directory.

-- sample_additive_scrambler_bb.json for public header file additive_scrambler_bb.h from gr-digital directory. \ No newline at end of file

-{

- "namespace": [

- "gr",

- "analog"

- ],

- "class": "agc2_cc",

- "io_signature": {

- "input": {

- "signature": "make",

- "min_streams": "1",

- "max_streams": "1",

- "sizeof_stream_item": "sizeof(gr_complex)"

- },

- "output": {

- "signature": "make",

- "min_streams": "1",

- "max_streams": "1",

- "sizeof_stream_item": "sizeof(gr_complex)"

- }

- },

- "message_port": {

- "input": [],

- "output": []

- },

- "make": {

- "arguments": [

- {

- "name": "attack_rate",

- "dtype": "float",

- "default": "0.10000000000000001"

- },

- {

- "name": "decay_rate",

- "dtype": "float",

- "default": "0.01"

- },

- {

- "name": "reference",

- "dtype": "float",

- "default": "1"

- },

- {

- "name": "gain",

- "dtype": "float",

- "default": "1"

- }

- ]

- },

- "methods": [

- {

- "name": "set_attack_rate",

- "arguments_type": [

- {

- "name": "rate",

- "dtype": "float"

- }

- ]

- },

- {

- "name": "set_decay_rate",

- "arguments_type": [

- {

- "name": "rate",

- "dtype": "float"

- }

- ]

- },

- {

- "name": "set_reference",

- "arguments_type": [

- {

- "name": "reference",

- "dtype": "float"

- }

- ]

- },

- {

- "name": "set_gain",

- "arguments_type": [

- {

- "name": "gain",

- "dtype": "float"

- }

- ]

- },

- {

- "name": "set_max_gain",

- "arguments_type": [

- {

- "name": "max_gain",

- "dtype": "float"

- }

- ]

- }

- ],

- "properties": [

- {

- "name": "attack_rate",

- "dtype": "float",

- "read_only": true

- },

- {

- "name": "decay_rate",

- "dtype": "float",

- "read_only": true

- },

- {

- "name": "reference",

- "dtype": "float",

- "read_only": false

- },

- {

- "name": "gain",

- "dtype": "float",

- "read_only": false

- },

- {

- "name": "max_gain",

- "dtype": "float",

- "read_only": false

- }

- ],

- "docstring": [

- "/*!\n * \\brief high performance Automatic Gain Control class with\n * attack and decay rates.\n * \\ingroup level_controllers_blk\n *\n * \\details\n * For Power the absolute value of the complex number is used.\n */",

- "// gr::analog::agc2_cc::sptr",

- "/*!\n * Build a complex value AGC loop block with attack and decay rates.\n *\n * \\param attack_rate the update rate of the loop when in attack mode.\n * \\param decay_rate the update rate of the loop when in decay mode.\n * \\param reference reference value to adjust signal power to.\n * \\param gain initial gain value.\n */",

- "/* namespace analog */",

- "/* namespace gr */",

- "/* INCLUDED_ANALOG_AGC2_CC_H */"

- ]

-} \ No newline at end of file

-{

- "namespace": [

- "gr",

- "digital"

- ],

- "class": "additive_scrambler_bb",

- "io_signature": {

- "input": {

- "signature": "make",

- "min_streams": "1",

- "max_streams": "1",

- "sizeof_stream_item": "sizeof(unsigned char)"

- },

- "output": {

- "signature": "make",

- "min_streams": "1",

- "max_streams": "1",

- "sizeof_stream_item": "sizeof(unsigned char)"

- }

- },

- "message_port": {

- "input": [],

- "output": []

- },

- "make": {

- "arguments": [

- {

- "name": "mask",

- "dtype": "int",

- "default": ""

- },

- {

- "name": "seed",

- "dtype": "int",

- "default": ""

- },

- {

- "name": "len",

- "dtype": "int",

- "default": ""

- },

- {

- "name": "count",

- "dtype": "int",

- "default": "0"

- },

- {

- "name": "bits_per_byte",

- "dtype": "int",

- "default": "1"

- },

- {

- "name": "reset_tag_key",

- "dtype": "std::string const &",

- "default": ""

- }

- ]

- },

- "methods": [],

- "properties": [

- {

- "name": "mask",

- "dtype": "int",

- "read_only": true

- },

- {

- "name": "seed",

- "dtype": "int",

- "read_only": true

- },

- {

- "name": "len",

- "dtype": "int",

- "read_only": true

- },

- {

- "name": "count",

- "dtype": "int",

- "read_only": true

- },

- {

- "name": "bits_per_byte",

- "dtype": "int",

- "read_only": true

- }

- ],

- "docstring": [

- "/*!\n * \\ingroup coding_blk\n *\n * \\brief\n * Scramble an input stream using an LFSR.\n *\n * \\details\n * This block scrambles up to 8 bits per byte of the input\n * data stream, starting at the LSB.\n *\n * The scrambler works by XORing the incoming bit stream by the\n * output of the LFSR. Optionally, after \\p count bits have been\n * processed, the shift register is reset to the \\p seed value.\n * This allows processing fixed length vectors of samples.\n *\n * Alternatively, the LFSR can be reset using a reset tag to\n * scramble variable length vectors. However, it cannot be reset\n * between bytes.\n *\n * For details on configuring the LFSR, see gr::digital::lfsr.\n */",

- "// gr::digital::additive_scrambler_bb::sptr",

- "/*!\n * \\brief Create additive scrambler.\n *\n * \\param mask Polynomial mask for LFSR\n * \\param seed Initial shift register contents\n * \\param len Shift register length\n * \\param count Number of bytes after which shift register is reset, 0=never\n * \\param bits_per_byte Number of bits per byte\n * \\param reset_tag_key When a tag with this key is detected, the shift register is reset (when this is set, count is ignored!)\n */",

- "/* namespace digital */",

- "/* namespace gr */",

- "/* INCLUDED_DIGITAL_ADDITIVE_SCRAMBLER_BB_H */"

- ]

-} \ No newline at end of file

-id: analog_agc2_cc

-label: AGC2

-category: '[Analog]'

-flags: '[python, cpp]'

-templates:

- imports: from gnuradio import analog

- make: analog.agc2_cc(${attack_rate}, ${decay_rate}, ${reference}, ${gain}, ${max_gain})

- callbacks: !!python/tuple

- - set_attack_rate(${rate})

- - set_decay_rate(${rate})

- - set_reference(${reference})

- - set_gain(${gain})

- - set_max_gain(${max_gain})

-parameters:

-- id: attack_rate

- label: Attack_rate

- dtype: float

- read_only: true

-- id: decay_rate

- label: Decay_rate

- dtype: float

- read_only: true

-- id: reference

- label: Reference

- dtype: float

- read_only: false

-- id: gain

- label: Gain

- dtype: float

- read_only: false

-- id: max_gain

- label: Max_gain

- dtype: float

- read_only: false

-inputs:

-- domain: stream

- dtype: sizeof(gr_complex)

-outputs:

-- domain: stream

- dtype: sizeof(gr_complex)

-cpp_templates:

- includes: '#include <gnuradio/analog/agc2_cc.h>'

- declartions: analog::agc2_cc::sptr ${id}

- make: this->${id} = analog::agc2_cc::make(${attack_rate}, ${decay_rate}, ${reference},

- ${gain}, ${max_gain})

- callbacks: !!python/tuple

- - set_attack_rate(${rate})

- - set_decay_rate(${rate})

- - set_reference(${reference})

- - set_gain(${gain})

- - set_max_gain(${max_gain})

- link: gnuradio-analog

-documentation:

-- "/*!\n * \\brief high performance Automatic Gain Control class with\n *

- attack and decay rates.\n * \\ingroup level_controllers_blk\n *\n *

- \\details\n * For Power the absolute value of the complex number is used.\n

- \ */"

-- // gr::analog::agc2_cc::sptr

-- "/*!\n * Build a complex value AGC loop block with attack and decay rates.\n

- \ *\n * \\param attack_rate the update rate of the loop when in attack

- mode.\n * \\param decay_rate the update rate of the loop when in decay mode.\n

- \ * \\param reference reference value to adjust signal power to.\n *

- \\param gain initial gain value.\n */"

-- /* namespace analog */

-- /* namespace gr */

-- /* INCLUDED_ANALOG_AGC2_CC_H */

-file_format: 1

-id: digital_additive_scrambler_bb

-label: ADDITIVE SCRAMBLER

-category: '[Digital]'

-flags: '[python, cpp]'

-templates:

- imports: from gnuradio import digital

- make: digital.additive_scrambler_bb(${mask}, ${seed}, ${len}, ${count}, ${bits_per_byte})

-parameters:

-- id: mask

- label: Mask

- dtype: int

- read_only: true

-- id: seed

- label: Seed

- dtype: int

- read_only: true

-- id: len

- label: Len

- dtype: int

- read_only: true

-- id: count

- label: Count

- dtype: int

- read_only: true

-- id: bits_per_byte

- label: Bits_per_byte

- dtype: int

- read_only: true

-inputs:

-- domain: stream

- dtype: sizeof(unsigned char)

-outputs:

-- domain: stream

- dtype: sizeof(unsigned char)

-cpp_templates:

- includes: '#include <gnuradio/digital/additive_scrambler_bb.h>'

- declartions: digital::additive_scrambler_bb::sptr ${id}

- make: this->${id} = digital::additive_scrambler_bb::make(${mask}, ${seed}, ${len},

- ${count}, ${bits_per_byte})

- link: gnuradio-digital

-documentation:

-- "/*!\n * \\ingroup coding_blk\n *\n * \\brief\n * Scramble an input

- stream using an LFSR.\n *\n * \\details\n * This block scrambles up

- to 8 bits per byte of the input\n * data stream, starting at the LSB.\n *\n

- \ * The scrambler works by XORing the incoming bit stream by the\n * output

- of the LFSR. Optionally, after \\p count bits have been\n * processed, the shift

- register is reset to the \\p seed value.\n * This allows processing fixed length

- vectors of samples.\n *\n * Alternatively, the LFSR can be reset using a

- reset tag to\n * scramble variable length vectors. However, it cannot be reset\n

- \ * between bytes.\n *\n * For details on configuring the LFSR, see gr::digital::lfsr.\n

- \ */"

-- // gr::digital::additive_scrambler_bb::sptr

-- "/*!\n * \\brief Create additive scrambler.\n *\n * \\param mask

- \ Polynomial mask for LFSR\n * \\param seed Initial shift register contents\n

- \ * \\param len Shift register length\n * \\param count Number of

- bytes after which shift register is reset, 0=never\n * \\param bits_per_byte

- Number of bits per byte\n * \\param reset_tag_key When a tag with this key

- is detected, the shift register is reset (when this is set, count is ignored!)\n

- \ */"

-- /* namespace digital */

-- /* namespace gr */

-- /* INCLUDED_DIGITAL_ADDITIVE_SCRAMBLER_BB_H */

-file_format: 1

-#

-# Copyright 2019 Free Software Foundation, Inc.

-#

-# This file is part of GNU Radio

-#

-# SPDX-License-Identifier: GPL-3.0-or-later

-#

-#

-""" unittest for gr-blocktool api """

-

-from __future__ import print_function

-from __future__ import absolute_import

-from __future__ import unicode_literals

-

-import os

-import unittest

-import warnings

-try:

- import pygccxml

- SKIP_BLOCK_TEST = False

-except:

- SKIP_BLOCK_TEST = True

-

-try:

- import apt

- CACHE = apt.cache.Cache()

- CACHE.open()

-

- PKG = CACHE['castxml']

- if PKG.is_installed:

- SKIP_BLOCK_TEST = False

- else:

- SKIP_BLOCK_TEST = True

-except:

- SKIP_BLOCK_TEST = True

-

-from jsonschema import validate

-

-from blocktool import BlockHeaderParser

-from blocktool.core.base import BlockToolException

-from blocktool.core import Constants

-from blocktool import RESULT_SCHEMA

-

-

-class TestBlocktoolCore(unittest.TestCase):

- """ The Tests for blocktool core """

-

- def __init__(self, *args, **kwargs):

- super(TestBlocktoolCore, self).__init__(*args, **kwargs)

- self.module = os.path.abspath(os.path.join(os.path.dirname(__file__),

- '../../../../gr-analog'))

- self.test_dir = os.path.abspath(os.path.join(self.module,

- 'include/gnuradio/analog'))

-

- def is_int(self, number):

- """

- Check for int conversion

- """

- try:

- int(number)

- return True

- except ValueError:

- return False

-

- @classmethod

- def setUpClass(cls):

- """ create a temporary Blocktool object """

- try:

- warnings.simplefilter("ignore", ResourceWarning)

- except NameError:

- pass

- test_path = {}

- target_file = os.path.abspath(os.path.join(os.path.dirname(

- __file__), '../../../../gr-analog/include/gnuradio/analog', 'agc2_cc.h'))

- test_path['file_path'] = target_file

- cls.test_obj = BlockHeaderParser(**test_path).get_header_info()

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_blocktool_exceptions(self):

- """

- tests for blocktool exceptions

- """

- # test for non-existent header or invalid headers

- test_dict = {}

- test_dict['file_path'] = os.path.abspath(

- os.path.join(self.test_dir, 'sample.h'))

- with self.assertRaises(BlockToolException):

- BlockHeaderParser(**test_dict).run_blocktool()

- # test for invalid header file

- test_dict['file_path'] = os.path.abspath(

- os.path.join(self.test_dir, 'CMakeLists.txt'))

- if not os.path.basename(test_dict['file_path']).endswith('.h'):

- with self.assertRaises(BlockToolException):

- BlockHeaderParser(**test_dict).run_blocktool()

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_namespace(self):

- """ test for header namespace """

- module_name = os.path.basename(self.module)

- self.assertTrue(self.test_obj['namespace'][0] == 'gr')

- self.assertTrue(self.test_obj['namespace']

- [1] == module_name.split('-')[-1])

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_io_signature(self):

- """ test for io_signature """

- input_signature = self.test_obj['io_signature']['input']['signature']

- output_signature = self.test_obj['io_signature']['output']['signature']

- valid_signature = False

- if input_signature and output_signature in Constants.SIGNATURE_LIST:

- valid_signature = True

- self.assertTrue(valid_signature)

- valid_io_stream = False

- input_max = self.test_obj['io_signature']['input']['max_streams']

- input_min = self.test_obj['io_signature']['input']['min_streams']

- output_max = self.test_obj['io_signature']['output']['max_streams']

- output_min = self.test_obj['io_signature']['output']['min_streams']

- if self.is_int(input_max) and self.is_int(input_min) and self.is_int(output_max) and self.is_int(output_min):

- valid_io_stream = True

- self.assertTrue(valid_io_stream)

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_message_port(self):

- """ test for message ports """

- input_port = self.test_obj['message_port']['input']

- output_port = self.test_obj['message_port']['output']

- valid_input_message_port = True

- valid_output_message_port = True

- if input_port:

- for port in input_port:

- if not port['id']:

- valid_input_message_port = False

- if output_port:

- for port in output_port:

- if not port['id']:

- valid_output_message_port = False

- self.assertTrue(valid_input_message_port)

- self.assertTrue(valid_output_message_port)

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_factory_signature(self):

- """ test for factory signature in the header """

- valid_factory_arg = True

- if self.test_obj['make']['arguments']:

- for arguments in self.test_obj['make']['arguments']:

- if not arguments['name'] or not arguments['dtype']:

- valid_factory_arg = False

- self.assertTrue(valid_factory_arg)

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_methods(self):

- """ test for methods """

- valid_method = True

- if self.test_obj['methods']:

- for arguments in self.test_obj['methods']:

- if not arguments['name']:

- valid_method = False

- if arguments['arguments_type']:

- for args in arguments['arguments_type']:

- if not args['name'] or not args['dtype']:

- valid_method = False

- self.assertTrue(valid_method)

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_properties(self):

- """ test for properties """

- valid_properties = True

- if self.test_obj['properties']:

- for arguments in self.test_obj['properties']:

- if not arguments['name'] or not arguments['dtype']:

- valid_properties = False

- self.assertTrue(valid_properties)

-

- @unittest.skipIf(SKIP_BLOCK_TEST, 'pygccxml not found, skipping this unittest')

- def test_result_format(self):

- """ test for parsed blocktool output format """

- valid_schema = False

- try:

- validate(instance=self.test_obj, schema=RESULT_SCHEMA)

- valid_schema = True

- except BlockToolException:

- raise BlockToolException

- self.assertTrue(valid_schema)

-

-

-if __name__ == '__main__':

- unittest.main()

-#

-# Copyright 2019 Free Software Foundation, Inc.

-#

-# This file is part of GNU Radio

-#

-# SPDX-License-Identifier: GPL-3.0-or-later

-#

-#

-""" testing the JSON files generated by gr-blocktool """

-

-from __future__ import print_function

-from __future__ import absolute_import

-from __future__ import unicode_literals

-

-import sys

-import json

-import jsonschema

-

-from blocktool import RESULT_SCHEMA

-

-

-def is_valid():

- """ Validate json file """

-

- with open(sys.argv[1], 'r') as json_file:

- data = json.load(json_file)

- try:

- print("Validating...")

- jsonschema.validate(data, RESULT_SCHEMA)

- except jsonschema.ValidationError as exception:

- print("Record JSON file # {}: NOT OK".format(sys.argv[1]))

- raise Exception(exception)

- else:

- print("Record JSON file # {}: OK".format(sys.argv[1]))

-

-

-if __name__ == '__main__':

- if len(sys.argv) == 2:

- is_valid()

- else:

- raise Exception('Please input only one json file')