gnuradio-companion/core/generator/hier_block.py

import collections
import os

import six
import codecs

from .top_block import TopBlockGenerator

from .. import Constants
from ..io import yaml


class HierBlockGenerator(TopBlockGenerator):
    """Extends the top block generator to also generate a block YML file"""

    def __init__(self, flow_graph, file_path):
        """
        Initialize the hier block generator object.

        Args:
            flow_graph: the flow graph object
            file_path: where to write the py file (the yml goes into HIER_BLOCK_LIB_DIR)
        """
        TopBlockGenerator.__init__(self, flow_graph, file_path)
        platform = flow_graph.parent

        hier_block_lib_dir = platform.config.hier_block_lib_dir
        if not os.path.exists(hier_block_lib_dir):
            os.mkdir(hier_block_lib_dir)

        self._mode = Constants.HIER_BLOCK_FILE_MODE
        self.file_path = os.path.join(hier_block_lib_dir, self._flow_graph.get_option('id') + '.py')
        self.file_path_yml = self.file_path + '.block.yml'

    def write(self):
        """generate output and write it to files"""
        TopBlockGenerator.write(self)

        data = yaml.dump(self._build_block_n_from_flow_graph_io())

        replace = [
            ('parameters:', '\nparameters:'),
            ('inputs:', '\ninputs:'),
            ('outputs:', '\noutputs:'),
            ('asserts:', '\nasserts:'),
            ('templates:', '\ntemplates:'),
            ('documentation:', '\ndocumentation:'),
            ('file_format:', '\nfile_format:'),
        ]
        for r in replace:
            data = data.replace(*r)

        with codecs.open(self.file_path_yml, 'w', encoding='utf-8') as fp:
            fp.write(data)

        # Windows only supports S_IREAD and S_IWRITE, other flags are ignored
        os.chmod(self.file_path_yml, self._mode)

    def _build_block_n_from_flow_graph_io(self):
        """
        Generate a block YML nested data from the flow graph IO

        Returns:
            a yml node tree
        """
        # Extract info from the flow graph
        block_id = self._flow_graph.get_option('id')
        parameters = self._flow_graph.get_parameters()

        def var_or_value(name):
            if name in (p.name for p in parameters):
                return "${" + name + " }"
            return name

        # Build the nested data
        data = collections.OrderedDict()
        data['id'] = block_id
        data['label'] = (
            self._flow_graph.get_option('title') or
            self._flow_graph.get_option('id').replace('_', ' ').title()
        )
        data['category'] = self._flow_graph.get_option('category')

        # Parameters
        data['parameters'] = []
        for param_block in parameters:
            p = collections.OrderedDict()
            p['id'] = param_block.name
            p['label'] = param_block.params['label'].get_value() or param_block.name
            p['dtype'] = param_block.params['value'].dtype
            p['default'] = param_block.params['value'].get_value()
            p['hide'] = param_block.params['hide'].get_value()
            data['parameters'].append(p)

        # Ports
        for direction in ('inputs', 'outputs'):
            data[direction] = []
            for port in get_hier_block_io(self._flow_graph, direction):
                p = collections.OrderedDict()
                if port.domain == Constants.GR_MESSAGE_DOMAIN:
                    p['id'] = port.key
                p['label'] = port.parent.params['label'].value
                if port.domain != Constants.DEFAULT_DOMAIN:
                    p['domain'] = port.domain
                p['dtype'] = port.dtype
                if port.domain != Constants.GR_MESSAGE_DOMAIN:
                    p['vlen'] = var_or_value(port.vlen)
                if port.optional:
                    p['optional'] = True
                data[direction].append(p)

        t = data['templates'] = collections.OrderedDict()

        t['imports'] = "from {0} import {0}  # grc-generated hier_block".format(
            self._flow_graph.get_option('id'))
        # Make data
        if parameters:
            t['make'] = '{cls}(\n    {kwargs},\n)'.format(
                cls=block_id,
                kwargs=',\n    '.join(
                    '{key}=${{ {key} }}'.format(key=param.name) for param in parameters
                ),
            )
        else:
            t['make'] = '{cls}()'.format(cls=block_id)
        # Callback data
        t['callbacks'] = [
            'set_{key}(${{ {key} }})'.format(key=param_block.name) for param_block in parameters
        ]


        # Documentation
        data['documentation'] = "\n".join(field for field in (
            self._flow_graph.get_option('author'),
            self._flow_graph.get_option('description'),
            self.file_path
        ) if field)
        data['grc_source'] = str(self._flow_graph.grc_file_path)

        data['file_format'] = 1

        return data


class QtHierBlockGenerator(HierBlockGenerator):

    def _build_block_n_from_flow_graph_io(self):
        n = HierBlockGenerator._build_block_n_from_flow_graph_io(self)
        block_n = collections.OrderedDict()

        # insert flags after category
        for key, value in six.iteritems(n):
            block_n[key] = value
            if key == 'category':
                block_n['flags'] = 'need_qt_gui'

        if not block_n['label'].upper().startswith('QT GUI'):
            block_n['label'] = 'QT GUI ' + block_n['label']

        gui_hint_param = collections.OrderedDict()
        gui_hint_param['id'] = 'gui_hint'
        gui_hint_param['label'] = 'GUI Hint'
        gui_hint_param['dtype'] = 'gui_hint'
        gui_hint_param['hide'] = 'part'
        block_n['parameters'].append(gui_hint_param)

        block_n['templates']['make'] += (
            "\n<% win = 'self.%s'%id %>"
            "\n${ gui_hint() % win }"
        )

        return block_n


def get_hier_block_io(flow_graph, direction, domain=None):
    """
    Get a list of io ports for this flow graph.

    Returns a list of blocks
    """
    pads = flow_graph.get_pad_sources() if direction == 'inputs' else flow_graph.get_pad_sinks()

    for pad in pads:
        for port in (pad.sources if direction == 'inputs' else pad.sinks):
            if domain and port.domain != domain:
                continue
            yield port