-"""

-Copyright 2008-2015 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

-"""

-

-import collections

-import itertools

-

-from Cheetah.Template import Template

-

-from .utils import epy_block_io, odict

-from . Constants import (

- BLOCK_FLAG_NEED_QT_GUI,

- ADVANCED_PARAM_TAB, DEFAULT_PARAM_TAB,

- BLOCK_FLAG_THROTTLE, BLOCK_FLAG_DISABLE_BYPASS,

- BLOCK_FLAG_DEPRECATED,

- BLOCK_ENABLED, BLOCK_BYPASSED, BLOCK_DISABLED

-)

-from . Element import Element

-

-

-def _get_keys(lst):

- return [elem.get_key() for elem in lst]

-

-

-def _get_elem(lst, key):

- try:

- return lst[_get_keys(lst).index(key)]

- except ValueError:

- raise ValueError('Key "{}" not found in {}.'.format(key, _get_keys(lst)))

-

-

-class Block(Element):

-

- is_block = True

-

- def __init__(self, flow_graph, n):

- """

- Make a new block from nested data.

-

- Args:

- flow: graph the parent element

- n: the nested odict

-

- Returns:

- block a new block

- """

- # Grab the data

- self._doc = (n.find('doc') or '').strip('\n').replace('\\\n', '')

- self._imports = map(lambda i: i.strip(), n.findall('import'))

- self._make = n.find('make')

- self._var_make = n.find('var_make')

- self._checks = n.findall('check')

- self._callbacks = n.findall('callback')

- self._bus_structure_source = n.find('bus_structure_source') or ''

- self._bus_structure_sink = n.find('bus_structure_sink') or ''

- self.port_counters = [itertools.count(), itertools.count()]

-

- # Build the block

- Element.__init__(self, flow_graph)

-

- # Grab the data

- params = n.findall('param')

- sources = n.findall('source')

- sinks = n.findall('sink')

- self._name = n.find('name')

- self._key = n.find('key')

- category = (n.find('category') or '').split('/')

- self.category = [cat.strip() for cat in category if cat.strip()]

- self._flags = n.find('flags') or ''

- # Backwards compatibility

- if n.find('throttle') and BLOCK_FLAG_THROTTLE not in self._flags:

- self._flags += BLOCK_FLAG_THROTTLE

- self._grc_source = n.find('grc_source') or ''

- self._block_wrapper_path = n.find('block_wrapper_path')

- self._bussify_sink = n.find('bus_sink')

- self._bussify_source = n.find('bus_source')

- self._var_value = n.find('var_value') or '$value'

-

- # Get list of param tabs

- n_tabs = n.find('param_tab_order') or None

- self._param_tab_labels = n_tabs.findall('tab') if n_tabs is not None else [DEFAULT_PARAM_TAB]

-

- # Create the param objects

- self._params = list()

-

- # Add the id param

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({

- 'name': 'ID',

- 'key': 'id',

- 'type': 'id',

- })

- ))

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({

- 'name': 'Enabled',

- 'key': '_enabled',

- 'type': 'raw',

- 'value': 'True',

- 'hide': 'all',

- })

- ))

- for param in itertools.imap(lambda n: self.get_parent().get_parent().Param(block=self, n=n), params):

- key = param.get_key()

- # Test against repeated keys

- if key in self.get_param_keys():

- raise Exception('Key "{}" already exists in params'.format(key))

- # Store the param

- self.get_params().append(param)

- # Create the source objects

- self._sources = list()

- for source in map(lambda n: self.get_parent().get_parent().Port(block=self, n=n, dir='source'), sources):

- key = source.get_key()

- # Test against repeated keys

- if key in self.get_source_keys():

- raise Exception('Key "{}" already exists in sources'.format(key))

- # Store the port

- self.get_sources().append(source)

- self.back_ofthe_bus(self.get_sources())

- # Create the sink objects

- self._sinks = list()

- for sink in map(lambda n: self.get_parent().get_parent().Port(block=self, n=n, dir='sink'), sinks):

- key = sink.get_key()

- # Test against repeated keys

- if key in self.get_sink_keys():

- raise Exception('Key "{}" already exists in sinks'.format(key))

- # Store the port

- self.get_sinks().append(sink)

- self.back_ofthe_bus(self.get_sinks())

- self.current_bus_structure = {'source': '', 'sink': ''}

-

- # Virtual source/sink and pad source/sink blocks are

- # indistinguishable from normal GR blocks. Make explicit

- # checks for them here since they have no work function or

- # buffers to manage.

- self.is_virtual_or_pad = self._key in (

- "virtual_source", "virtual_sink", "pad_source", "pad_sink")

- self.is_variable = self._key.startswith('variable')

- self.is_import = (self._key == 'import')

-

- # Disable blocks that are virtual/pads or variables

- if self.is_virtual_or_pad or self.is_variable:

- self._flags += BLOCK_FLAG_DISABLE_BYPASS

-

- if not (self.is_virtual_or_pad or self.is_variable or self._key == 'options'):

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({'name': 'Block Alias',

- 'key': 'alias',

- 'type': 'string',

- 'hide': 'part',

- 'tab': ADVANCED_PARAM_TAB

- })

- ))

-

- if (len(sources) or len(sinks)) and not self.is_virtual_or_pad:

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({'name': 'Core Affinity',

- 'key': 'affinity',

- 'type': 'int_vector',

- 'hide': 'part',

- 'tab': ADVANCED_PARAM_TAB

- })

- ))

- if len(sources) and not self.is_virtual_or_pad:

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({'name': 'Min Output Buffer',

- 'key': 'minoutbuf',

- 'type': 'int',

- 'hide': 'part',

- 'value': '0',

- 'tab': ADVANCED_PARAM_TAB

- })

- ))

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({'name': 'Max Output Buffer',

- 'key': 'maxoutbuf',

- 'type': 'int',

- 'hide': 'part',

- 'value': '0',

- 'tab': ADVANCED_PARAM_TAB

- })

- ))

-

- self.get_params().append(self.get_parent().get_parent().Param(

- block=self,

- n=odict({'name': 'Comment',

- 'key': 'comment',

- 'type': '_multiline',

- 'hide': 'part',

- 'value': '',

- 'tab': ADVANCED_PARAM_TAB

- })

- ))

-

- self._epy_source_hash = -1 # for epy blocks

- self._epy_reload_error = None

-

- if self._bussify_sink:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'sink')

- if self._bussify_source:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'source')

-

- def get_bus_structure(self, direction):

- if direction == 'source':

- bus_structure = self._bus_structure_source

- else:

- bus_structure = self._bus_structure_sink

-

- bus_structure = self.resolve_dependencies(bus_structure)

-

- if not bus_structure:

- return '' # TODO: Don't like empty strings. should change this to None eventually

-

- try:

- clean_bus_structure = self.get_parent().evaluate(bus_structure)

- return clean_bus_structure

- except:

- return ''

-

- def validate(self):

- """

- Validate this block.

- Call the base class validate.

- Evaluate the checks: each check must evaluate to True.

- """

- Element.validate(self)

- # Evaluate the checks

- for check in self._checks:

- check_res = self.resolve_dependencies(check)

- try:

- if not self.get_parent().evaluate(check_res):

- self.add_error_message('Check "{}" failed.'.format(check))

- except:

- self.add_error_message('Check "{}" did not evaluate.'.format(check))

-

- # For variables check the value (only if var_value is used

- if self.is_variable and self._var_value != '$value':

- value = self._var_value

- try:

- value = self.get_var_value()

- self.get_parent().evaluate(value)

- except Exception as err:

- self.add_error_message('Value "{}" cannot be evaluated:\n{}'.format(value, err))

-

- # check if this is a GUI block and matches the selected generate option

- current_generate_option = self.get_parent().get_option('generate_options')

-

- def check_generate_mode(label, flag, valid_options):

- block_requires_mode = (

- flag in self.get_flags() or

- self.get_name().upper().startswith(label)

- )

- if block_requires_mode and current_generate_option not in valid_options:

- self.add_error_message("Can't generate this block in mode: {} ".format(

- repr(current_generate_option)))

-

- check_generate_mode('QT GUI', BLOCK_FLAG_NEED_QT_GUI, ('qt_gui', 'hb_qt_gui'))

- if self._epy_reload_error:

- self.get_param('_source_code').add_error_message(str(self._epy_reload_error))

-

- def rewrite(self):

- """

- Add and remove ports to adjust for the nports.

- """

- Element.rewrite(self)

- # Check and run any custom rewrite function for this block

- getattr(self, 'rewrite_' + self._key, lambda: None)()

-

- # Adjust nports, disconnect hidden ports

- for ports in (self.get_sources(), self.get_sinks()):

- for i, master_port in enumerate(ports):

- nports = master_port.get_nports() or 1

- num_ports = 1 + len(master_port.get_clones())

- if master_port.get_hide():

- for connection in master_port.get_connections():

- self.get_parent().remove_element(connection)

- if not nports and num_ports == 1: # Not a master port and no left-over clones

- continue

- # Remove excess cloned ports

- for port in master_port.get_clones()[nports-1:]:

- # Remove excess connections

- for connection in port.get_connections():

- self.get_parent().remove_element(connection)

- master_port.remove_clone(port)

- ports.remove(port)

- # Add more cloned ports

- for j in range(num_ports, nports):

- port = master_port.add_clone()

- ports.insert(ports.index(master_port) + j, port)

-

- self.back_ofthe_bus(ports)

- # Renumber non-message/message ports

- domain_specific_port_index = collections.defaultdict(int)

- for port in filter(lambda p: p.get_key().isdigit(), ports):

- domain = port.get_domain()

- port._key = str(domain_specific_port_index[domain])

- domain_specific_port_index[domain] += 1

-

- def port_controller_modify(self, direction):

- """

- Change the port controller.

-

- Args:

- direction: +1 or -1

-

- Returns:

- true for change

- """

- changed = False

- # Concat the nports string from the private nports settings of all ports

- nports_str = ' '.join([port._nports for port in self.get_ports()])

- # Modify all params whose keys appear in the nports string

- for param in self.get_params():

- if param.is_enum() or param.get_key() not in nports_str:

- continue

- # Try to increment the port controller by direction

- try:

- value = param.get_evaluated()

- value = value + direction

- if 0 < value:

- param.set_value(value)

- changed = True

- except:

- pass

- return changed

-

- def get_doc(self):

- platform = self.get_parent().get_parent()

- documentation = platform.block_docstrings.get(self._key, {})

- from_xml = self._doc.strip()

- if from_xml:

- documentation[''] = from_xml

- return documentation

-

- def get_imports(self, raw=False):

- """

- Resolve all import statements.

- Split each import statement at newlines.

- Combine all import statements into a list.

- Filter empty imports.

-

- Returns:

- a list of import statements

- """

- if raw:

- return self._imports

- return filter(lambda i: i, sum(map(lambda i: self.resolve_dependencies(i).split('\n'), self._imports), []))

-

- def get_make(self, raw=False):

- if raw:

- return self._make

- return self.resolve_dependencies(self._make)

-

- def get_var_make(self):

- return self.resolve_dependencies(self._var_make)

-

- def get_var_value(self):

- return self.resolve_dependencies(self._var_value)

-

- def get_callbacks(self):

- """

- Get a list of function callbacks for this block.

-

- Returns:

- a list of strings

- """

- def make_callback(callback):

- callback = self.resolve_dependencies(callback)

- if 'self.' in callback:

- return callback

- return 'self.{}.{}'.format(self.get_id(), callback)

- return map(make_callback, self._callbacks)

-

- def is_virtual_sink(self):

- return self.get_key() == 'virtual_sink'

-

- def is_virtual_source(self):

- return self.get_key() == 'virtual_source'

-

- ###########################################################################

- # Custom rewrite functions

- ###########################################################################

-

- def rewrite_epy_block(self):

- flowgraph = self.get_parent()

- platform = flowgraph.get_parent()

- param_blk = self.get_param('_io_cache')

- param_src = self.get_param('_source_code')

-

- src = param_src.get_value()

- src_hash = hash((self.get_id(), src))

- if src_hash == self._epy_source_hash:

- return

-

- try:

- blk_io = epy_block_io.extract(src)

-

- except Exception as e:

- self._epy_reload_error = ValueError(str(e))

- try: # Load last working block io

- blk_io_args = eval(param_blk.get_value())

- if len(blk_io_args) == 6:

- blk_io_args += ([],) # add empty callbacks

- blk_io = epy_block_io.BlockIO(*blk_io_args)

- except Exception:

- return

- else:

- self._epy_reload_error = None # Clear previous errors

- param_blk.set_value(repr(tuple(blk_io)))

-

- # print "Rewriting embedded python block {!r}".format(self.get_id())

-

- self._epy_source_hash = src_hash

- self._name = blk_io.name or blk_io.cls

- self._doc = blk_io.doc

- self._imports[0] = 'import ' + self.get_id()

- self._make = '{0}.{1}({2})'.format(self.get_id(), blk_io.cls, ', '.join(

- '{0}=${{ {0} }}'.format(key) for key, _ in blk_io.params))

- self._callbacks = ['{0} = ${{ {0} }}'.format(attr) for attr in blk_io.callbacks]

-

- params = {}

- for param in list(self._params):

- if hasattr(param, '__epy_param__'):

- params[param.get_key()] = param

- self._params.remove(param)

-

- for key, value in blk_io.params:

- try:

- param = params[key]

- param.set_default(value)

- except KeyError: # need to make a new param

- name = key.replace('_', ' ').title()

- n = odict(dict(name=name, key=key, type='raw', value=value))

- param = platform.Param(block=self, n=n)

- setattr(param, '__epy_param__', True)

- self._params.append(param)

-

- def update_ports(label, ports, port_specs, direction):

- ports_to_remove = list(ports)

- iter_ports = iter(ports)

- ports_new = []

- port_current = next(iter_ports, None)

- for key, port_type, vlen in port_specs:

- reuse_port = (

- port_current is not None and

- port_current.get_type() == port_type and

- port_current.get_vlen() == vlen and

- (key.isdigit() or port_current.get_key() == key)

- )

- if reuse_port:

- ports_to_remove.remove(port_current)

- port, port_current = port_current, next(iter_ports, None)

- else:

- n = odict(dict(name=label + str(key), type=port_type, key=key))

- if port_type == 'message':

- n['name'] = key

- n['optional'] = '1'

- if vlen > 1:

- n['vlen'] = str(vlen)

- port = platform.Port(block=self, n=n, dir=direction)

- ports_new.append(port)

- # replace old port list with new one

- del ports[:]

- ports.extend(ports_new)

- # remove excess port connections

- for port in ports_to_remove:

- for connection in port.get_connections():

- flowgraph.remove_element(connection)

-

- update_ports('in', self.get_sinks(), blk_io.sinks, 'sink')

- update_ports('out', self.get_sources(), blk_io.sources, 'source')

- self.rewrite()

-

- def back_ofthe_bus(self, portlist):

- portlist.sort(key=lambda p: p._type == 'bus')

-

- def filter_bus_port(self, ports):

- buslist = [p for p in ports if p._type == 'bus']

- return buslist or ports

-

- # Main functions to get and set the block state

- # Also kept get_enabled and set_enabled to keep compatibility

- def get_state(self):

- """

- Gets the block's current state.

-

- Returns:

- ENABLED - 0

- BYPASSED - 1

- DISABLED - 2

- """

- try:

- return int(eval(self.get_param('_enabled').get_value()))

- except:

- return BLOCK_ENABLED

-

- def set_state(self, state):

- """

- Sets the state for the block.

-

- Args:

- ENABLED - 0

- BYPASSED - 1

- DISABLED - 2

- """

- if state in [BLOCK_ENABLED, BLOCK_BYPASSED, BLOCK_DISABLED]:

- self.get_param('_enabled').set_value(str(state))

- else:

- self.get_param('_enabled').set_value(str(BLOCK_ENABLED))

-

- # Enable/Disable Aliases

- def get_enabled(self):

- """

- Get the enabled state of the block.

-

- Returns:

- true for enabled

- """

- return not (self.get_state() == BLOCK_DISABLED)

-

- def set_enabled(self, enabled):

- """

- Set the enabled state of the block.

-

- Args:

- enabled: true for enabled

-

- Returns:

- True if block changed state

- """

- old_state = self.get_state()

- new_state = BLOCK_ENABLED if enabled else BLOCK_DISABLED

- self.set_state(new_state)

- return old_state != new_state

-

- # Block bypassing

- def get_bypassed(self):

- """

- Check if the block is bypassed

- """

- return self.get_state() == BLOCK_BYPASSED

-

- def set_bypassed(self):

- """

- Bypass the block

-

- Returns:

- True if block chagnes state

- """

- if self.get_state() != BLOCK_BYPASSED and self.can_bypass():

- self.set_state(BLOCK_BYPASSED)

- return True

- return False

-

- def can_bypass(self):

- """ Check the number of sinks and sources and see if this block can be bypassed """

- # Check to make sure this is a single path block

- # Could possibly support 1 to many blocks

- if len(self.get_sources()) != 1 or len(self.get_sinks()) != 1:

- return False

- if not (self.get_sources()[0].get_type() == self.get_sinks()[0].get_type()):

- return False

- if self.bypass_disabled():

- return False

- return True

-

- def __str__(self):

- return 'Block - {} - {}({})'.format(self.get_id(), self.get_name(), self.get_key())

-

- def get_id(self):

- return self.get_param('id').get_value()

-

- def get_name(self):

- return self._name

-

- def get_key(self):

- return self._key

-

- def get_ports(self):

- return self.get_sources() + self.get_sinks()

-

- def get_ports_gui(self):

- return self.filter_bus_port(self.get_sources()) + self.filter_bus_port(self.get_sinks())

-

- def get_children(self):

- return self.get_ports() + self.get_params()

-

- def get_children_gui(self):

- return self.get_ports_gui() + self.get_params()

-

- def get_block_wrapper_path(self):

- return self._block_wrapper_path

-

- def get_comment(self):

- return self.get_param('comment').get_value()

-

- def get_flags(self):

- return self._flags

-

- def throtteling(self):

- return BLOCK_FLAG_THROTTLE in self._flags

-

- def bypass_disabled(self):

- return BLOCK_FLAG_DISABLE_BYPASS in self._flags

-

- @property

- def is_deprecated(self):

- return BLOCK_FLAG_DEPRECATED in self._flags

-

- ##############################################

- # Access Params

- ##############################################

- def get_param_tab_labels(self):

- return self._param_tab_labels

-

- def get_param_keys(self):

- return _get_keys(self._params)

-

- def get_param(self, key):

- return _get_elem(self._params, key)

-

- def get_params(self):

- return self._params

-

- def has_param(self, key):

- try:

- _get_elem(self._params, key)

- return True

- except:

- return False

-

- ##############################################

- # Access Sinks

- ##############################################

- def get_sink_keys(self):

- return _get_keys(self._sinks)

-

- def get_sink(self, key):

- return _get_elem(self._sinks, key)

-

- def get_sinks(self):

- return self._sinks

-

- def get_sinks_gui(self):

- return self.filter_bus_port(self.get_sinks())

-

- ##############################################

- # Access Sources

- ##############################################

- def get_source_keys(self):

- return _get_keys(self._sources)

-

- def get_source(self, key):

- return _get_elem(self._sources, key)

-

- def get_sources(self):

- return self._sources

-

- def get_sources_gui(self):

- return self.filter_bus_port(self.get_sources())

-

- def get_connections(self):

- return sum([port.get_connections() for port in self.get_ports()], [])

-

- def resolve_dependencies(self, tmpl):

- """

- Resolve a parameter dependency with cheetah templates.

-

- Args:

- tmpl: the string with dependencies

-

- Returns:

- the resolved value

- """

- tmpl = str(tmpl)

- if '$' not in tmpl:

- return tmpl

- n = dict((param.get_key(), param.template_arg)

- for param in self.get_params()) # TODO: cache that

- try:

- return str(Template(tmpl, n))

- except Exception as err:

- return "Template error: {}\n {}".format(tmpl, err)

-

- ##############################################

- # Controller Modify

- ##############################################

- def type_controller_modify(self, direction):

- """

- Change the type controller.

-

- Args:

- direction: +1 or -1

-

- Returns:

- true for change

- """

- changed = False

- type_param = None

- for param in filter(lambda p: p.is_enum(), self.get_params()):

- children = self.get_ports() + self.get_params()

- # Priority to the type controller

- if param.get_key() in ' '.join(map(lambda p: p._type, children)): type_param = param

- # Use param if type param is unset

- if not type_param:

- type_param = param

- if type_param:

- # Try to increment the enum by direction

- try:

- keys = type_param.get_option_keys()

- old_index = keys.index(type_param.get_value())

- new_index = (old_index + direction + len(keys)) % len(keys)

- type_param.set_value(keys[new_index])

- changed = True

- except:

- pass

- return changed

-

- def form_bus_structure(self, direc):

- if direc == 'source':

- get_p = self.get_sources

- get_p_gui = self.get_sources_gui

- bus_structure = self.get_bus_structure('source')

- else:

- get_p = self.get_sinks

- get_p_gui = self.get_sinks_gui

- bus_structure = self.get_bus_structure('sink')

-

- struct = [range(len(get_p()))]

- if True in map(lambda a: isinstance(a.get_nports(), int), get_p()):

- structlet = []

- last = 0

- for j in [i.get_nports() for i in get_p() if isinstance(i.get_nports(), int)]:

- structlet.extend(map(lambda a: a+last, range(j)))

- last = structlet[-1] + 1

- struct = [structlet]

- if bus_structure:

-

- struct = bus_structure

-

- self.current_bus_structure[direc] = struct

- return struct

-

- def bussify(self, n, direc):

- if direc == 'source':

- get_p = self.get_sources

- get_p_gui = self.get_sources_gui

- bus_structure = self.get_bus_structure('source')

- else:

- get_p = self.get_sinks

- get_p_gui = self.get_sinks_gui

- bus_structure = self.get_bus_structure('sink')

-

- for elt in get_p():

- for connect in elt.get_connections():

- self.get_parent().remove_element(connect)

-

- if ('bus' not in map(lambda a: a.get_type(), get_p())) and len(get_p()) > 0:

- struct = self.form_bus_structure(direc)

- self.current_bus_structure[direc] = struct

- if get_p()[0].get_nports():

- n['nports'] = str(1)

-

- for i in range(len(struct)):

- n['key'] = str(len(get_p()))

- n = odict(n)

- port = self.get_parent().get_parent().Port(block=self, n=n, dir=direc)

- get_p().append(port)

- elif 'bus' in map(lambda a: a.get_type(), get_p()):

- for elt in get_p_gui():

- get_p().remove(elt)

- self.current_bus_structure[direc] = ''

-

- ##############################################

- # Import/Export Methods

- ##############################################

- def export_data(self):

- """

- Export this block's params to nested data.

-

- Returns:

- a nested data odict

- """

- n = odict()

- n['key'] = self.get_key()

- n['param'] = map(lambda p: p.export_data(), sorted(self.get_params(), key=str))

- if 'bus' in map(lambda a: a.get_type(), self.get_sinks()):

- n['bus_sink'] = str(1)

- if 'bus' in map(lambda a: a.get_type(), self.get_sources()):

- n['bus_source'] = str(1)

- return n

-

- def get_hash(self):

- return hash(tuple(map(hash, self.get_params())))

-

- def import_data(self, n):

- """

- Import this block's params from nested data.

- Any param keys that do not exist will be ignored.

- Since params can be dynamically created based another param,

- call rewrite, and repeat the load until the params stick.

- This call to rewrite will also create any dynamic ports

- that are needed for the connections creation phase.

-

- Args:

- n: the nested data odict

- """

- my_hash = 0

- while self.get_hash() != my_hash:

- params_n = n.findall('param')

- for param_n in params_n:

- key = param_n.find('key')

- value = param_n.find('value')

- # The key must exist in this block's params

- if key in self.get_param_keys():

- self.get_param(key).set_value(value)

- # Store hash and call rewrite

- my_hash = self.get_hash()

- self.rewrite()

- bussinks = n.findall('bus_sink')

- if len(bussinks) > 0 and not self._bussify_sink:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'sink')

- elif len(bussinks) > 0:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'sink')

- self.bussify({'name': 'bus', 'type': 'bus'}, 'sink')

- bussrcs = n.findall('bus_source')

- if len(bussrcs) > 0 and not self._bussify_source:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'source')

- elif len(bussrcs) > 0:

- self.bussify({'name': 'bus', 'type': 'bus'}, 'source')

- self.bussify({'name': 'bus', 'type': 'bus'}, 'source')

-# Copyright 2011 Free Software Foundation, Inc.

-#

-# This file is part of GNU Radio

-#

-# GNU Radio is free software; you can redistribute it and/or modify

-# it under the terms of the GNU General Public License as published by

-# the Free Software Foundation; either version 3, or (at your option)

-# any later version.

-#

-# GNU Radio is distributed in the hope that it will be useful,

-# but WITHOUT ANY WARRANTY; without even the implied warranty of

-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-# GNU General Public License for more details.

-#

-# You should have received a copy of the GNU General Public License

-# along with GNU Radio; see the file COPYING. If not, write to

-# the Free Software Foundation, Inc., 51 Franklin Street,

-# Boston, MA 02110-1301, USA.

-

-file(GLOB py_files "*.py")

-

-GR_PYTHON_INSTALL(

- FILES ${py_files}

- DESTINATION ${GR_PYTHON_DIR}/gnuradio/grc/core

-)

-

-file(GLOB dtd_files "*.dtd")

-

-install(

- FILES ${dtd_files} default_flow_graph.grc

- DESTINATION ${GR_PYTHON_DIR}/gnuradio/grc/core

-)

-

-add_subdirectory(generator)

-add_subdirectory(utils)

-"""

-Copyright 2016 Free Software Foundation, Inc.

-

- key = 'grc'

- def __init__(self, prefs_file, version, version_parts=None, name=None):

- self.prefs = prefs_file

- self.prefs.get_string('grc', 'local_blocks_path', ''),

- self.prefs.get_string('grc', 'global_blocks_path', ''),

- self.prefs.get_string('grc', 'default_flow_graph', '') or

-from . import Constants

-from .Element import Element

-from .utils import odict

- def __init__(self, flow_graph, porta, portb):

- porta: a port (any direction)

- portb: a port (any direction)

- Element.__init__(self, flow_graph)

- source = sink = None

- # Separate the source and sink

- for port in (porta, portb):

- if port.is_source:

- source = port

- else:

- sink = port

- if not source:

- if not sink:

- busses = len(filter(lambda a: a.get_type() == 'bus', [source, sink])) % 2

- if not busses == 0:

- raise ValueError('busses must get with busses')

-

- if not len(source.get_associated_ports()) == len(sink.get_associated_ports()):

- raise ValueError('port connections must have same cardinality')

- # Ensure that this connection (source -> sink) is unique

- for connection in flow_graph.connections:

- if connection.get_source() is source and connection.get_sink() is sink:

- raise LookupError('This connection between source and sink is not unique.')

- self._source = source

- self._sink = sink

- if source.get_type() == 'bus':

-

- sources = source.get_associated_ports()

- sinks = sink.get_associated_ports()

-

- for i in range(len(sources)):

- try:

- flow_graph.connect(sources[i], sinks[i])

- except:

- pass

- self.get_source().get_parent(),

- self.get_source(),

- self.get_sink().get_parent(),

- self.get_sink(),

- def is_bus(self):

- return self.get_source().get_type() == self.get_sink().get_type() == 'bus'

- def validate(self):

- """

- Validate the connections.

- The ports must match in io size.

- """

- """

- Validate the connections.

- The ports must match in type.

- """

- Element.validate(self)

- platform = self.get_parent().get_parent()

- source_domain = self.get_source().get_domain()

- sink_domain = self.get_sink().get_domain()

- if (source_domain, sink_domain) not in platform.connection_templates:

- self.add_error_message('No connection known for domains "{}", "{}"'.format(

- source_domain, sink_domain))

- too_many_other_sinks = (

- not platform.domains.get(source_domain, []).get('multiple_sinks', False) and

- len(self.get_source().get_enabled_connections()) > 1

- )

- too_many_other_sources = (

- not platform.domains.get(sink_domain, []).get('multiple_sources', False) and

- len(self.get_sink().get_enabled_connections()) > 1

- )

- if too_many_other_sinks:

- self.add_error_message(

- 'Domain "{}" can have only one downstream block'.format(source_domain))

- if too_many_other_sources:

- self.add_error_message(

- 'Domain "{}" can have only one upstream block'.format(sink_domain))

-

- source_size = Constants.TYPE_TO_SIZEOF[self.get_source().get_type()] * self.get_source().get_vlen()

- sink_size = Constants.TYPE_TO_SIZEOF[self.get_sink().get_type()] * self.get_sink().get_vlen()

- if source_size != sink_size:

- self.add_error_message('Source IO size "{}" does not match sink IO size "{}".'.format(source_size, sink_size))

- def get_enabled(self):

- return self.get_source().get_parent().get_enabled() and \

- self.get_sink().get_parent().get_enabled()

- #############################

- # Access Ports

- #############################

- def get_sink(self):

- return self._sink

- def get_source(self):

- return self._source

- n = odict()

- n['source_block_id'] = self.get_source().get_parent().get_id()

- n['sink_block_id'] = self.get_sink().get_parent().get_id()

- n['source_key'] = self.get_source().get_key()

- n['sink_key'] = self.get_sink().get_key()

- return n

-import numpy

-FLOW_GRAPH_DTD = os.path.join(DATA_DIR, 'flow_graph.dtd')

-BLOCK_TREE_DTD = os.path.join(DATA_DIR, 'block_tree.dtd')

-DOMAIN_DTD = os.path.join(DATA_DIR, 'domain.dtd')

-GR_STREAM_DOMAIN = "gr_stream"

-GR_MESSAGE_DOMAIN = "gr_message"

-BLOCK_FLAG_THROTTLE = 'throttle'

-BLOCK_FLAG_DISABLE_BYPASS = 'disable_bypass'

-BLOCK_FLAG_NEED_QT_GUI = 'need_qt_gui'

-BLOCK_FLAG_DEPRECATED = 'deprecated'

-

-# Block States

-BLOCK_DISABLED = 0

-BLOCK_ENABLED = 1

-BLOCK_BYPASSED = 2

-

-COMPLEX_TYPES = [complex, numpy.complex, numpy.complex64, numpy.complex128]

-REAL_TYPES = [float, numpy.float, numpy.float32, numpy.float64]

-INT_TYPES = [int, long, numpy.int, numpy.int8, numpy.int16, numpy.int32, numpy.uint64,

- numpy.uint, numpy.uint8, numpy.uint16, numpy.uint32, numpy.uint64]

-# Cast to tuple for isinstance, concat subtypes

-COMPLEX_TYPES = tuple(COMPLEX_TYPES + REAL_TYPES + INT_TYPES)

-REAL_TYPES = tuple(REAL_TYPES + INT_TYPES)

-INT_TYPES = tuple(INT_TYPES)

- ('Complex Float 64', 'fc64', 16, GRC_COLOR_BROWN),

- ('Complex Float 32', 'fc32', 8, GRC_COLOR_BLUE),

- ('Complex Integer 64', 'sc64', 16, GRC_COLOR_LIGHT_GREEN),

- ('Complex Integer 32', 'sc32', 8, GRC_COLOR_GREEN),

- ('Complex Integer 16', 'sc16', 4, GRC_COLOR_AMBER),

- ('Complex Integer 8', 'sc8', 2, GRC_COLOR_PURPLE),

- ('Float 64', 'f64', 8, GRC_COLOR_CYAN),

- ('Float 32', 'f32', 4, GRC_COLOR_ORANGE),

- ('Integer 64', 's64', 8, GRC_COLOR_LIME),

- ('Integer 32', 's32', 4, GRC_COLOR_TEAL),

- ('Integer 16', 's16', 2, GRC_COLOR_YELLOW),

- ('Integer 8', 's8', 1, GRC_COLOR_PURPLE_A400),

- ('Bits (unpacked byte)', 'bit', 1, GRC_COLOR_PURPLE_A100),

- ('Async Message', 'message', 0, GRC_COLOR_GREY),

- ('Bus Connection', 'bus', 0, GRC_COLOR_WHITE),

- ('Wildcard', '', 0, GRC_COLOR_WHITE),

- 'float': (4, GRC_COLOR_ORANGE),

- 'int': (4, GRC_COLOR_TEAL),

- 'short': (2, GRC_COLOR_YELLOW),

- 'byte': (1, GRC_COLOR_PURPLE_A400),

- 'bits': (1, GRC_COLOR_PURPLE_A100),

-TYPE_TO_COLOR = dict()

-TYPE_TO_SIZEOF = dict()

-

-for name, key, sizeof, color in CORE_TYPES:

- TYPE_TO_COLOR[key] = color

- TYPE_TO_SIZEOF[key] = sizeof

-

-for key, (sizeof, color) in ALIAS_TYPES.iteritems():

- TYPE_TO_COLOR[key] = color

- TYPE_TO_SIZEOF[key] = sizeof

-

-# Coloring

-COMPLEX_COLOR_SPEC = '#3399FF'

-FLOAT_COLOR_SPEC = '#FF8C69'

-INT_COLOR_SPEC = '#00FF99'

-SHORT_COLOR_SPEC = '#FFFF66'

-BYTE_COLOR_SPEC = '#FF66FF'

-COMPLEX_VECTOR_COLOR_SPEC = '#3399AA'

-FLOAT_VECTOR_COLOR_SPEC = '#CC8C69'

-INT_VECTOR_COLOR_SPEC = '#00CC99'

-SHORT_VECTOR_COLOR_SPEC = '#CCCC33'

-BYTE_VECTOR_COLOR_SPEC = '#CC66CC'

-ID_COLOR_SPEC = '#DDDDDD'

-WILDCARD_COLOR_SPEC = '#FFFFFF'

-"""

-Copyright 2008, 2009, 2015 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

-"""

-

-

-class Element(object):

-

- def __init__(self, parent=None):

- self._parent = parent

- self._error_messages = list()

-

- ##################################################

- # Element Validation API

- ##################################################

- def validate(self):

- """

- Validate this element and call validate on all children.

- Call this base method before adding error messages in the subclass.

- """

- del self._error_messages[:]

- for child in self.get_children():

- child.validate()

-

- def is_valid(self):

- """

- Is this element valid?

-

- Returns:

- true when the element is enabled and has no error messages or is bypassed

- """

- return (not self.get_error_messages() or not self.get_enabled()) or self.get_bypassed()

-

- def add_error_message(self, msg):

- """

- Add an error message to the list of errors.

-

- Args:

- msg: the error message string

- """

- self._error_messages.append(msg)

-

- def get_error_messages(self):

- """

- Get the list of error messages from this element and all of its children.

- Do not include the error messages from disabled or bypassed children.

- Cleverly indent the children error messages for printing purposes.

-

- Returns:

- a list of error message strings

- """

- error_messages = list(self._error_messages) # Make a copy

- for child in filter(lambda c: c.get_enabled() and not c.get_bypassed(), self.get_children()):

- for msg in child.get_error_messages():

- error_messages.append("{}:\n\t{}".format(child, msg.replace("\n", "\n\t")))

- return error_messages

-

- def rewrite(self):

- """

- Rewrite this element and call rewrite on all children.

- Call this base method before rewriting the element.

- """

- for child in self.get_children():

- child.rewrite()

-

- def get_enabled(self):

- return True

-

- def get_bypassed(self):

- return False

-

- ##############################################

- # Tree-like API

- ##############################################

- def get_parent(self):

- return self._parent

-

- def get_children(self):

- return list()

-

- ##############################################

- # Type testing

- ##############################################

- is_platform = False

-

- is_flow_graph = False

-

- is_block = False

-

- is_dummy_block = False

-

- is_connection = False

-

- is_port = False

-

- is_param = False

-

- is_variable = False

-

- is_import = False

-from itertools import ifilter, chain

-from operator import methodcaller, attrgetter

-import re

-import time

-from . import Messages

-from .Element import Element

-from .utils import odict, expr_utils, shlex

-

-_parameter_matcher = re.compile('^(parameter)$')

-_monitors_searcher = re.compile('(ctrlport_monitor)')

-_bussink_searcher = re.compile('^(bus_sink)$')

-_bussrc_searcher = re.compile('^(bus_source)$')

-_bus_struct_sink_searcher = re.compile('^(bus_structure_sink)$')

-_bus_struct_src_searcher = re.compile('^(bus_structure_source)$')

- def __init__(self, platform):

- platform: a platforms with blocks and contrcutors

- Element.__init__(self, platform)

- self._elements = []

- self._timestamp = time.ctime()

- self.platform = platform # todo: make this a lazy prop

- self.blocks = []

- self.connections = []

- self._options_block = self.new_block('options')

- ##############################################

- # TODO: Move these to new generator package

- ##############################################

- def get_imports(self):

- a set of import statements

- imports = sum([block.get_imports() for block in self.iter_enabled_blocks()], [])

- return sorted(set(imports))

- variables = filter(attrgetter('is_variable'), self.iter_enabled_blocks())

- return expr_utils.sort_objects(variables, methodcaller('get_id'), methodcaller('get_var_make'))

- parameters = filter(lambda b: _parameter_matcher.match(b.get_key()), self.iter_enabled_blocks())

- monitors = filter(lambda b: _monitors_searcher.search(b.get_key()),

- self.iter_enabled_blocks())

- if block.get_key() == 'epy_module':

- yield block.get_id(), block.get_param('source_code').get_value()

-

- def get_bussink(self):

- bussink = filter(lambda b: _bussink_searcher.search(b.get_key()), self.get_enabled_blocks())

-

- for i in bussink:

- for j in i.get_params():

- if j.get_name() == 'On/Off' and j.get_value() == 'on':

- return True

- return False

-

- def get_bussrc(self):

- bussrc = filter(lambda b: _bussrc_searcher.search(b.get_key()), self.get_enabled_blocks())

-

- for i in bussrc:

- for j in i.get_params():

- if j.get_name() == 'On/Off' and j.get_value() == 'on':

- return True

- return False

-

- def get_bus_structure_sink(self):

- bussink = filter(lambda b: _bus_struct_sink_searcher.search(b.get_key()), self.get_enabled_blocks())

- return bussink

-

- def get_bus_structure_src(self):

- bussrc = filter(lambda b: _bus_struct_src_searcher.search(b.get_key()), self.get_enabled_blocks())

- return bussrc

- return ifilter(methodcaller('get_enabled'), self.blocks)

- return filter(methodcaller('get_bypassed'), self.blocks)

- return filter(methodcaller('get_enabled'), self.connections)

- return self._options_block.get_param(key).get_evaluated()

- def get_block(self, id):

- if block.get_id() == id:

- raise KeyError('No block with ID {!r}'.format(id))

- """

- Get a list of all the elements.

- Always ensure that the options block is in the list (only once).

- Returns:

- the element list

- """

- options_block_count = self.blocks.count(self._options_block)

- if not options_block_count:

- self.blocks.append(self._options_block)

- for i in range(options_block_count-1):

- self.blocks.remove(self._options_block)

-

- return self.blocks + self.connections

-

- get_children = get_elements

-

- for child in chain(self.blocks, self.connections):

- child.rewrite()

-

- self.bus_ports_rewrite()

- for expr in self.get_imports():

- exec expr in namespace

- exec expr in module.__dict__

- for parameter in self.get_parameters():

- value = eval(parameter.get_param('value').to_code(), namespace)

- np[parameter.get_id()] = value

- for variable in self.get_variables():

- value = eval(variable.get_var_value(), namespace)

- namespace[variable.get_id()] = value

- def evaluate(self, expr):

-

- Args:

- expr: the string expression

- @throw Exception bad expression

-

- Returns:

- the evaluated data

- return self._eval_cache.setdefault(expr, eval(expr, self.namespace))

- def new_block(self, key):

- key: the block key

- block = self.platform.get_new_block(self, key)

-

- connection = self.platform.Connection(

- flow_graph=self, porta=porta, portb=portb)

- self.connections.append(connection)

- # Found a port, set to parent signal block

- element = element.get_parent()

- for port in element.get_ports():

- map(self.remove_element, port.get_connections())

- if element.is_bus():

- cons_list = []

- for i in map(lambda a: a.get_connections(), element.get_source().get_associated_ports()):

- cons_list.extend(i)

- map(self.remove_element, cons_list)

- # sort blocks and connections for nicer diffs

- blocks = sorted(self.blocks, key=lambda b: (

- b.get_key() != 'options', # options to the front

- not b.get_key().startswith('variable'), # then vars

- str(b)

- ))

- connections = sorted(self.connections, key=str)

- n = odict()

- n['timestamp'] = self._timestamp

- n['block'] = [b.export_data() for b in blocks]

- n['connection'] = [c.export_data() for c in connections]

- instructions = odict({

- 'created': '.'.join(self.get_parent().config.version_parts),

- 'format': FLOW_GRAPH_FILE_FORMAT_VERSION,

- })

- return odict({'flow_graph': n, '_instructions': instructions})

-

- def import_data(self, n):

- Clear this flowgraph of all previous blocks and connections.

- n: the nested data odict

- del self.connections[:]

- # set file format

- try:

- instructions = n.find('_instructions') or {}

- file_format = int(instructions.get('format', '0')) or _guess_file_format_1(n)

- except:

- file_format = 0

- fg_n = n and n.find('flow_graph') or odict() # use blank data if none provided

- self._timestamp = fg_n.find('timestamp') or time.ctime()

- self._options_block = self.new_block('options')

- for block_n in fg_n.findall('block'):

- key = block_n.find('key')

- block = self._options_block if key == 'options' else self.new_block(key)

-

- if not block:

- # we're before the initial fg update(), so no evaluated values!

- # --> use raw value instead

- path_param = self._options_block.get_param('hier_block_src_path')

- file_path = self.platform.find_file_in_paths(

- filename=key + '.grc',

- paths=path_param.get_value(),

- cwd=self.grc_file_path

- )

- if file_path: # grc file found. load and get block

- self.platform.load_and_generate_flow_graph(file_path, hier_only=True)

- block = self.new_block(key) # can be None

-

- if not block: # looks like this block key cannot be found

- # create a dummy block instead

- block = self.new_block('dummy_block')

- # Ugly ugly ugly

- _initialize_dummy_block(block, block_n)

- print('Block key "%s" not found' % key)

-

- block.import_data(block_n)

- ports = block.get_sinks() if dir == 'sink' else block.get_sources()

- if key == port.get_key():

- if not key.isdigit() and port.get_type() == '' and key == port.get_name():

- port = _dummy_block_add_port(block, key, dir)

- errors = False

- for connection_n in fg_n.findall('connection'):

- # get the block ids and port keys

- source_block_id = connection_n.find('source_block_id')

- sink_block_id = connection_n.find('sink_block_id')

- source_key = connection_n.find('source_key')

- sink_key = connection_n.find('sink_key')

- try:

- source_block = self.get_block(source_block_id)

- sink_block = self.get_block(sink_block_id)

- source_key, sink_key = _update_old_message_port_keys(

- source_key, sink_key, source_block, sink_block)

- source_port = verify_and_get_port(source_key, source_block, 'source')

- sink_port = verify_and_get_port(sink_key, sink_block, 'sink')

- except LookupError as e:

- Messages.send_error_load(

- 'Connection between {}({}) and {}({}) could not be made.\n\t{}'.format(

- source_block_id, source_key, sink_block_id, sink_key, e))

- errors = True

- self.rewrite() # global rewrite

- return errors

- ##############################################

- # Needs to go

- ##############################################

- def bus_ports_rewrite(self):

- # todo: move to block.rewrite()

- for block in self.blocks:

- for direc in ['source', 'sink']:

- if direc == 'source':

- get_p = block.get_sources

- get_p_gui = block.get_sources_gui

- bus_structure = block.form_bus_structure('source')

- else:

- get_p = block.get_sinks

- get_p_gui = block.get_sinks_gui

- bus_structure = block.form_bus_structure('sink')

-

- if 'bus' in map(lambda a: a.get_type(), get_p_gui()):

- if len(get_p_gui()) > len(bus_structure):

- times = range(len(bus_structure), len(get_p_gui()))

- for i in times:

- for connect in get_p_gui()[-1].get_connections():

- block.get_parent().remove_element(connect)

- get_p().remove(get_p_gui()[-1])

- elif len(get_p_gui()) < len(bus_structure):

- n = {'name': 'bus', 'type': 'bus'}

- if True in map(

- lambda a: isinstance(a.get_nports(), int),

- get_p()):

- n['nports'] = str(1)

-

- times = range(len(get_p_gui()), len(bus_structure))

-

- for i in times:

- n['key'] = str(len(get_p()))

- n = odict(n)

- port = block.get_parent().get_parent().Port(

- block=block, n=n, dir=direc)

- get_p().append(port)

-

- if 'bus' in map(lambda a: a.get_type(),

- block.get_sources_gui()):

- for i in range(len(block.get_sources_gui())):

- if len(block.get_sources_gui()[

- i].get_connections()) > 0:

- source = block.get_sources_gui()[i]

- sink = []

-

- for j in range(len(source.get_connections())):

- sink.append(

- source.get_connections()[j].get_sink())

- for elt in source.get_connections():

- self.remove_element(elt)

- for j in sink:

- self.connect(source, j)

- # get ports using the "old way" (assuming liner indexed keys)

- source_port = source_block.get_sources()[int(source_key)]

- sink_port = sink_block.get_sinks()[int(sink_key)]

- if source_port.get_type() == "message" and sink_port.get_type() == "message":

- source_key, sink_key = source_port.get_key(), sink_port.get_key()

-

-

-def _guess_file_format_1(n):

- """

- Try to guess the file format for flow-graph files without version tag

- """

- try:

- has_non_numeric_message_keys = any(not (

- connection_n.find('source_key').isdigit() and

- connection_n.find('sink_key').isdigit()

- ) for connection_n in n.find('flow_graph').findall('connection'))

- if has_non_numeric_message_keys:

- return 1

- except:

- pass

- return 0

-

-

-def _initialize_dummy_block(block, block_n):

- """

- This is so ugly... dummy-fy a block

- Modify block object to get the behaviour for a missing block

- """

-

- block._key = block_n.find('key')

- block.is_dummy_block = lambda: True

- block.is_valid = lambda: False

- block.get_enabled = lambda: False

- for param_n in block_n.findall('param'):

- if param_n['key'] not in block.get_param_keys():

- new_param_n = odict({'key': param_n['key'], 'name': param_n['key'], 'type': 'string'})

- params = block.get_parent().get_parent().Param(block=block, n=new_param_n)

- block.get_params().append(params)

-

-

-def _dummy_block_add_port(block, key, dir):

- """ This is so ugly... Add a port to a dummy-field block """

- port_n = odict({'name': '?', 'key': key, 'type': ''})

- port = block.get_parent().get_parent().Port(block=block, n=port_n, dir=dir)

- if port.is_source:

- block.get_sources().append(port)

- else:

- block.get_sinks().append(port)

- return port

-import os

-def send_fail_connection():

- send('>>> Error: Cannot create connection.\n')

-"""

-Copyright 2008-2015 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

-"""

-

-import ast

-import weakref

-import re

-

-from . import Constants

-from .Constants import VECTOR_TYPES, COMPLEX_TYPES, REAL_TYPES, INT_TYPES

-from .Element import Element

-from .utils import odict

-

-# Blacklist certain ids, its not complete, but should help

-import __builtin__

-

-

-ID_BLACKLIST = ['self', 'options', 'gr', 'math', 'firdes'] + dir(__builtin__)

-try:

- from gnuradio import gr

- ID_BLACKLIST.extend(attr for attr in dir(gr.top_block()) if not attr.startswith('_'))

-except ImportError:

- pass

-

-_check_id_matcher = re.compile('^[a-z|A-Z]\w*$')

-_show_id_matcher = re.compile('^(variable\w*|parameter|options|notebook)$')

-

-

-def _get_keys(lst):

- return [elem.get_key() for elem in lst]

-

-

-def _get_elem(lst, key):

- try:

- return lst[_get_keys(lst).index(key)]

- except ValueError:

- raise ValueError('Key "{}" not found in {}.'.format(key, _get_keys(lst)))

-

-

-def num_to_str(num):

- """ Display logic for numbers """

- def eng_notation(value, fmt='g'):

- """Convert a number to a string in engineering notation. E.g., 5e-9 -> 5n"""

- template = '{:' + fmt + '}{}'

- magnitude = abs(value)

- for exp, symbol in zip(range(9, -15-1, -3), 'GMk munpf'):

- factor = 10 ** exp

- if magnitude >= factor:

- return template.format(value / factor, symbol.strip())

- return template.format(value, '')

-

- if isinstance(num, COMPLEX_TYPES):

- num = complex(num) # Cast to python complex

- if num == 0:

- return '0'

- output = eng_notation(num.real) if num.real else ''

- output += eng_notation(num.imag, '+g' if output else 'g') + 'j' if num.imag else ''

- return output

- else:

- return str(num)

-

-

-class Option(Element):

-

- def __init__(self, param, n):

- Element.__init__(self, param)

- self._name = n.find('name')

- self._key = n.find('key')

- self._opts = dict()

- opts = n.findall('opt')

- # Test against opts when non enum

- if not self.get_parent().is_enum() and opts:

- raise Exception('Options for non-enum types cannot have sub-options')

- # Extract opts

- for opt in opts:

- # Separate the key:value

- try:

- key, value = opt.split(':')

- except:

- raise Exception('Error separating "{}" into key:value'.format(opt))

- # Test against repeated keys

- if key in self._opts:

- raise Exception('Key "{}" already exists in option'.format(key))

- # Store the option

- self._opts[key] = value

-

- def __str__(self):

- return 'Option {}({})'.format(self.get_name(), self.get_key())

-

- def get_name(self):

- return self._name

-

- def get_key(self):

- return self._key

-

- ##############################################

- # Access Opts

- ##############################################

- def get_opt_keys(self):

- return self._opts.keys()

-

- def get_opt(self, key):

- return self._opts[key]

-

- def get_opts(self):

- return self._opts.values()

-

-

-class TemplateArg(object):

- """

- A cheetah template argument created from a param.

- The str of this class evaluates to the param's to code method.

- The use of this class as a dictionary (enum only) will reveal the enum opts.

- The __call__ or () method can return the param evaluated to a raw python data type.

- """

-

- def __init__(self, param):

- self._param = weakref.proxy(param)

-

- def __getitem__(self, item):

- return str(self._param.get_opt(item)) if self._param.is_enum() else NotImplemented

-

- def __getattr__(self, item):

- if not self._param.is_enum():

- raise AttributeError()

- try:

- return str(self._param.get_opt(item))

- except KeyError:

- raise AttributeError()

-

- def __str__(self):

- return str(self._param.to_code())

-

- def __call__(self):

- return self._param.get_evaluated()

-

-

-class Param(Element):

-

- is_param = True

-

- def __init__(self, block, n):

- """

- Make a new param from nested data.

-

- Args:

- block: the parent element

- n: the nested odict

- """

- # If the base key is a valid param key, copy its data and overlay this params data

- base_key = n.find('base_key')

- if base_key and base_key in block.get_param_keys():

- n_expanded = block.get_param(base_key)._n.copy()

- n_expanded.update(n)

- n = n_expanded

- # Save odict in case this param will be base for another

- self._n = n

- # Parse the data

- self._name = n.find('name')

- self._key = n.find('key')

- value = n.find('value') or ''

- self._type = n.find('type') or 'raw'

- self._hide = n.find('hide') or ''

- self._tab_label = n.find('tab') or block.get_param_tab_labels()[0]

- if self._tab_label not in block.get_param_tab_labels():

- block.get_param_tab_labels().append(self._tab_label)

- # Build the param

- Element.__init__(self, block)

- # Create the Option objects from the n data

- self._options = list()

- self._evaluated = None

- for option in map(lambda o: Option(param=self, n=o), n.findall('option')):

- key = option.get_key()

- # Test against repeated keys

- if key in self.get_option_keys():

- raise Exception('Key "{}" already exists in options'.format(key))

- # Store the option

- self.get_options().append(option)

- # Test the enum options

- if self.is_enum():

- # Test against options with identical keys

- if len(set(self.get_option_keys())) != len(self.get_options()):

- raise Exception('Options keys "{}" are not unique.'.format(self.get_option_keys()))

- # Test against inconsistent keys in options

- opt_keys = self.get_options()[0].get_opt_keys()

- for option in self.get_options():

- if set(opt_keys) != set(option.get_opt_keys()):

- raise Exception('Opt keys "{}" are not identical across all options.'.format(opt_keys))

- # If a value is specified, it must be in the options keys

- if value or value in self.get_option_keys():

- self._value = value

- else:

- self._value = self.get_option_keys()[0]

- if self.get_value() not in self.get_option_keys():

- raise Exception('The value "{}" is not in the possible values of "{}".'.format(self.get_value(), self.get_option_keys()))

- else:

- self._value = value or ''

- self._default = value

- self._init = False

- self._hostage_cells = list()

- self.template_arg = TemplateArg(self)

-

- def get_types(self):

- return (

- 'raw', 'enum',

- 'complex', 'real', 'float', 'int',

- 'complex_vector', 'real_vector', 'float_vector', 'int_vector',

- 'hex', 'string', 'bool',

- 'file_open', 'file_save', '_multiline', '_multiline_python_external',

- 'id', 'stream_id',

- 'gui_hint',

- 'import',

- )

-

- def __repr__(self):

- """

- Get the repr (nice string format) for this param.

-

- Returns:

- the string representation

- """

- ##################################################

- # Truncate helper method

- ##################################################

- def _truncate(string, style=0):

- max_len = max(27 - len(self.get_name()), 3)

- if len(string) > max_len:

- if style < 0: # Front truncate

- string = '...' + string[3-max_len:]

- elif style == 0: # Center truncate

- string = string[:max_len/2 - 3] + '...' + string[-max_len/2:]

- elif style > 0: # Rear truncate

- string = string[:max_len-3] + '...'

- return string

-

- ##################################################

- # Simple conditions

- ##################################################

- if not self.is_valid():

- return _truncate(self.get_value())

- if self.get_value() in self.get_option_keys():

- return self.get_option(self.get_value()).get_name()

-

- ##################################################

- # Split up formatting by type

- ##################################################

- # Default center truncate

- truncate = 0

- e = self.get_evaluated()

- t = self.get_type()

- if isinstance(e, bool):

- return str(e)

- elif isinstance(e, COMPLEX_TYPES):

- dt_str = num_to_str(e)

- elif isinstance(e, VECTOR_TYPES):

- # Vector types

- if len(e) > 8:

- # Large vectors use code

- dt_str = self.get_value()

- truncate = 1

- else:

- # Small vectors use eval

- dt_str = ', '.join(map(num_to_str, e))

- elif t in ('file_open', 'file_save'):

- dt_str = self.get_value()

- truncate = -1

- else:

- # Other types

- dt_str = str(e)

-

- # Done

- return _truncate(dt_str, truncate)

-

- def __repr2__(self):

- """

- Get the repr (nice string format) for this param.

-

- Returns:

- the string representation

- """

- if self.is_enum():

- return self.get_option(self.get_value()).get_name()

- return self.get_value()

-

- def __str__(self):

- return 'Param - {}({})'.format(self.get_name(), self.get_key())

-

- def get_color(self):

- """

- Get the color that represents this param's type.

-

- Returns:

- a hex color code.

- """

- try:

- return {

- # Number types

- 'complex': Constants.COMPLEX_COLOR_SPEC,

- 'real': Constants.FLOAT_COLOR_SPEC,

- 'float': Constants.FLOAT_COLOR_SPEC,

- 'int': Constants.INT_COLOR_SPEC,

- # Vector types

- 'complex_vector': Constants.COMPLEX_VECTOR_COLOR_SPEC,

- 'real_vector': Constants.FLOAT_VECTOR_COLOR_SPEC,

- 'float_vector': Constants.FLOAT_VECTOR_COLOR_SPEC,

- 'int_vector': Constants.INT_VECTOR_COLOR_SPEC,

- # Special

- 'bool': Constants.INT_COLOR_SPEC,

- 'hex': Constants.INT_COLOR_SPEC,

- 'string': Constants.BYTE_VECTOR_COLOR_SPEC,

- 'id': Constants.ID_COLOR_SPEC,

- 'stream_id': Constants.ID_COLOR_SPEC,

- 'raw': Constants.WILDCARD_COLOR_SPEC,

- }[self.get_type()]

- except:

- return '#FFFFFF'

-

- def get_hide(self):

- """

- Get the hide value from the base class.

- Hide the ID parameter for most blocks. Exceptions below.

- If the parameter controls a port type, vlen, or nports, return part.

- If the parameter is an empty grid position, return part.

- These parameters are redundant to display in the flow graph view.

-

- Returns:

- hide the hide property string

- """

- hide = self.get_parent().resolve_dependencies(self._hide).strip()

- if hide:

- return hide

- # Hide ID in non variable blocks

- if self.get_key() == 'id' and not _show_id_matcher.match(self.get_parent().get_key()):

- return 'part'

- # Hide port controllers for type and nports

- if self.get_key() in ' '.join(map(lambda p: ' '.join([p._type, p._nports]),

- self.get_parent().get_ports())):

- return 'part'

- # Hide port controllers for vlen, when == 1

- if self.get_key() in ' '.join(map(

- lambda p: p._vlen, self.get_parent().get_ports())

- ):

- try:

- if int(self.get_evaluated()) == 1:

- return 'part'

- except:

- pass

- return hide

-

- def validate(self):

- """

- Validate the param.

- The value must be evaluated and type must a possible type.

- """

- Element.validate(self)

- if self.get_type() not in self.get_types():

- self.add_error_message('Type "{}" is not a possible type.'.format(self.get_type()))

-

- self._evaluated = None

- try:

- self._evaluated = self.evaluate()

- except Exception, e:

- self.add_error_message(str(e))

-

- def get_evaluated(self):

- return self._evaluated

-

- def evaluate(self):

- """

- Evaluate the value.

-

- Returns:

- evaluated type

- """

- self._init = True

- self._lisitify_flag = False

- self._stringify_flag = False

- self._hostage_cells = list()

- t = self.get_type()

- v = self.get_value()

-

- #########################

- # Enum Type

- #########################

- if self.is_enum():

- return v

-

- #########################

- # Numeric Types

- #########################

- elif t in ('raw', 'complex', 'real', 'float', 'int', 'hex', 'bool'):

- # Raise exception if python cannot evaluate this value

- try:

- e = self.get_parent().get_parent().evaluate(v)

- except Exception, e:

- raise Exception('Value "{}" cannot be evaluated:\n{}'.format(v, e))

- # Raise an exception if the data is invalid

- if t == 'raw':

- return e

- elif t == 'complex':

- if not isinstance(e, COMPLEX_TYPES):

- raise Exception('Expression "{}" is invalid for type complex.'.format(str(e)))

- return e

- elif t == 'real' or t == 'float':

- if not isinstance(e, REAL_TYPES):

- raise Exception('Expression "{}" is invalid for type float.'.format(str(e)))

- return e

- elif t == 'int':

- if not isinstance(e, INT_TYPES):

- raise Exception('Expression "{}" is invalid for type integer.'.format(str(e)))

- return e

- elif t == 'hex':

- return hex(e)

- elif t == 'bool':

- if not isinstance(e, bool):

- raise Exception('Expression "{}" is invalid for type bool.'.format(str(e)))

- return e

- else:

- raise TypeError('Type "{}" not handled'.format(t))

- #########################

- # Numeric Vector Types

- #########################

- elif t in ('complex_vector', 'real_vector', 'float_vector', 'int_vector'):

- if not v:

- # Turn a blank string into an empty list, so it will eval

- v = '()'

- # Raise exception if python cannot evaluate this value

- try:

- e = self.get_parent().get_parent().evaluate(v)

- except Exception, e:

- raise Exception('Value "{}" cannot be evaluated:\n{}'.format(v, e))

- # Raise an exception if the data is invalid

- if t == 'complex_vector':

- if not isinstance(e, VECTOR_TYPES):

- self._lisitify_flag = True

- e = [e]

- if not all([isinstance(ei, COMPLEX_TYPES) for ei in e]):

- raise Exception('Expression "{}" is invalid for type complex vector.'.format(str(e)))

- return e

- elif t == 'real_vector' or t == 'float_vector':

- if not isinstance(e, VECTOR_TYPES):

- self._lisitify_flag = True

- e = [e]

- if not all([isinstance(ei, REAL_TYPES) for ei in e]):

- raise Exception('Expression "{}" is invalid for type float vector.'.format(str(e)))

- return e

- elif t == 'int_vector':

- if not isinstance(e, VECTOR_TYPES):

- self._lisitify_flag = True

- e = [e]

- if not all([isinstance(ei, INT_TYPES) for ei in e]):

- raise Exception('Expression "{}" is invalid for type integer vector.'.format(str(e)))

- return e

- #########################

- # String Types

- #########################

- elif t in ('string', 'file_open', 'file_save', '_multiline', '_multiline_python_external'):

- # Do not check if file/directory exists, that is a runtime issue

- try:

- e = self.get_parent().get_parent().evaluate(v)

- if not isinstance(e, str):

- raise Exception()

- except:

- self._stringify_flag = True

- e = str(v)

- if t == '_multiline_python_external':

- ast.parse(e) # Raises SyntaxError

- return e

- #########################

- # Unique ID Type

- #########################

- elif t == 'id':

- # Can python use this as a variable?

- if not _check_id_matcher.match(v):

- raise Exception('ID "{}" must begin with a letter and may contain letters, numbers, and underscores.'.format(v))

- ids = [param.get_value() for param in self.get_all_params(t, 'id')]

-

- if v in ID_BLACKLIST:

- raise Exception('ID "{}" is blacklisted.'.format(v))

-

- if self._key == 'id':

- # Id should only appear once, or zero times if block is disabled

- if ids.count(v) > 1:

- raise Exception('ID "{}" is not unique.'.format(v))

- else:

- # Id should exist to be a reference

- if ids.count(v) < 1:

- raise Exception('ID "{}" does not exist.'.format(v))

-

- return v

-

- #########################

- # Stream ID Type

- #########################

- elif t == 'stream_id':

- # Get a list of all stream ids used in the virtual sinks

- ids = [param.get_value() for param in filter(

- lambda p: p.get_parent().is_virtual_sink(),

- self.get_all_params(t),

- )]

- # Check that the virtual sink's stream id is unique

- if self.get_parent().is_virtual_sink():

- # Id should only appear once, or zero times if block is disabled

- if ids.count(v) > 1:

- raise Exception('Stream ID "{}" is not unique.'.format(v))

- # Check that the virtual source's steam id is found

- if self.get_parent().is_virtual_source():

- if v not in ids:

- raise Exception('Stream ID "{}" is not found.'.format(v))

- return v

-

- #########################

- # GUI Position/Hint

- #########################

- elif t == 'gui_hint':

- if ':' in v:

- tab, pos = v.split(':')

- elif '@' in v:

- tab, pos = v, ''

- else:

- tab, pos = '', v

-

- if '@' in tab:

- tab, index = tab.split('@')

- else:

- index = '?'

-

- # TODO: Problem with this code. Produces bad tabs

- widget_str = ({

- (True, True): 'self.%(tab)s_grid_layout_%(index)s.addWidget(%(widget)s, %(pos)s)',

- (True, False): 'self.%(tab)s_layout_%(index)s.addWidget(%(widget)s)',

- (False, True): 'self.top_grid_layout.addWidget(%(widget)s, %(pos)s)',

- (False, False): 'self.top_layout.addWidget(%(widget)s)',

- }[bool(tab), bool(pos)]) % {'tab': tab, 'index': index, 'widget': '%s', 'pos': pos}

-

- # FIXME: Move replace(...) into the make template of the qtgui blocks

- # Return a string here

- class GuiHint(object):

- def __init__(self, ws):

- self._ws = ws

-

- def __call__(self, w):

- return (self._ws.replace('addWidget', 'addLayout') if 'layout' in w else self._ws) % w

-

- def __str__(self):

- return self._ws

- return GuiHint(widget_str)

- #########################

- # Import Type

- #########################

- elif t == 'import':

- # New namespace

- n = dict()

- try:

- exec v in n

- except ImportError:

- raise Exception('Import "{}" failed.'.format(v))

- except Exception:

- raise Exception('Bad import syntax: "{}".'.format(v))

- return filter(lambda k: str(k) != '__builtins__', n.keys())

-

- #########################

- else:

- raise TypeError('Type "{}" not handled'.format(t))

-

- def to_code(self):

- """

- Convert the value to code.

- For string and list types, check the init flag, call evaluate().

- This ensures that evaluate() was called to set the xxxify_flags.

-

- Returns:

- a string representing the code

- """

- v = self.get_value()

- t = self.get_type()

- # String types

- if t in ('string', 'file_open', 'file_save', '_multiline', '_multiline_python_external'):

- if not self._init:

- self.evaluate()

- return repr(v) if self._stringify_flag else v

-

- # Vector types

- elif t in ('complex_vector', 'real_vector', 'float_vector', 'int_vector'):

- if not self._init:

- self.evaluate()

- if self._lisitify_flag:

- return '(%s, )' % v

- else:

- return '(%s)' % v

- else:

- return v

-

- def get_all_params(self, type, key=None):

- """

- Get all the params from the flowgraph that have the given type and

- optionally a given key

-

- Args:

- type: the specified type

- key: the key to match against

-

- Returns:

- a list of params

- """

- return sum([filter(lambda p: ((p.get_type() == type) and ((key is None) or (p.get_key() == key))), block.get_params()) for block in self.get_parent().get_parent().get_enabled_blocks()], [])

-

- def is_enum(self):

- return self._type == 'enum'

-

- def get_value(self):

- value = self._value

- if self.is_enum() and value not in self.get_option_keys():

- value = self.get_option_keys()[0]

- self.set_value(value)

- return value

-

- def set_value(self, value):

- # Must be a string

- self._value = str(value)

-

- def set_default(self, value):

- if self._default == self._value:

- self.set_value(value)

- self._default = str(value)

-

- def get_type(self):

- return self.get_parent().resolve_dependencies(self._type)

-

- def get_tab_label(self):

- return self._tab_label

-

- def get_name(self):

- return self.get_parent().resolve_dependencies(self._name).strip()

-

- def get_key(self):

- return self._key

-

- ##############################################

- # Access Options

- ##############################################

- def get_option_keys(self):

- return _get_keys(self.get_options())

-

- def get_option(self, key):

- return _get_elem(self.get_options(), key)

-

- def get_options(self):

- return self._options

-

- ##############################################

- # Access Opts

- ##############################################

- def get_opt_keys(self):

- return self.get_option(self.get_value()).get_opt_keys()

-

- def get_opt(self, key):

- return self.get_option(self.get_value()).get_opt(key)

-

- def get_opts(self):

- return self.get_option(self.get_value()).get_opts()

-

- ##############################################

- # Import/Export Methods

- ##############################################

- def export_data(self):

- """

- Export this param's key/value.

-

- Returns:

- a nested data odict

- """

- n = odict()

- n['key'] = self.get_key()

- n['value'] = self.get_value()

- return n

-from .utils import odict

- nested_data = _from_file(xml.getroot())

- nested_data['_instructions'] = {}

- for inst in filter(lambda i: i.target == 'grc', xml_instructions):

- nested_data['_instructions'] = odict(inst.attrib)

-def _from_file(xml):

- return odict({tag: xml.text or ''}) # store empty tags (text is None) as empty string

- nested_data = odict()

- key, value = _from_file(elem).items()[0]

- if key in nested_data:

- nested_data[key].append(value)

- nested_data[key] = [value]

- # Delistify if the length of values is 1

- for key, values in nested_data.iteritems():

- if len(values) == 1:

- nested_data[key] = values[0]

- return odict({tag: nested_data})

- "{0}='{1}'".format(*item) for item in instructions.iteritems())

- with open(xml_file, 'w') as fp:

- for key, values in nested_data.iteritems():

- if isinstance(value, (str, unicode)):

- node.text = unicode(value)

-"""

-Copyright 2008-2016 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

-"""

-

-import os

-import sys

-

-from . import ParseXML, Messages, Constants

-

-from .Config import Config

-from .Element import Element

-from .generator import Generator

-from .FlowGraph import FlowGraph

-from .Connection import Connection

-from .Block import Block

-from .Port import Port

-from .Param import Param

-

-from .utils import odict, extract_docs, hide_bokeh_gui_options_if_not_installed

-

-

-class Platform(Element):

-

- Config = Config

- Generator = Generator

- FlowGraph = FlowGraph

- Connection = Connection

- Block = Block

- Port = Port

- Param = Param

-

- is_platform = True

-

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

- """ Make a platform for GNU Radio """

- Element.__init__(self)

-

- self.config = self.Config(*args, **kwargs)

-

- self.block_docstrings = {}

- self.block_docstrings_loaded_callback = lambda: None # dummy to be replaced by BlockTreeWindow

-

- self._docstring_extractor = extract_docs.SubprocessLoader(

- callback_query_result=self._save_docstring_extraction_result,

- callback_finished=lambda: self.block_docstrings_loaded_callback()

- )

-

- # Create a dummy flow graph for the blocks

- self._flow_graph = Element(self)

- self._flow_graph.connections = []

-

- self.blocks = odict()

- self._blocks_n = odict()

- self._block_categories = {}

- self.domains = {}

- self.connection_templates = {}

-

- self._auto_hier_block_generate_chain = set()

-

- self.build_block_library()

-

- def __str__(self):

- return 'Platform - {}({})'.format(self.config.key, self.config.name)

-

- @staticmethod

- def find_file_in_paths(filename, paths, cwd):

- """Checks the provided paths relative to cwd for a certain filename"""

- if not os.path.isdir(cwd):

- cwd = os.path.dirname(cwd)

- if isinstance(paths, str):

- paths = (p for p in paths.split(':') if p)

-

- for path in paths:

- path = os.path.expanduser(path)

- if not os.path.isabs(path):

- path = os.path.normpath(os.path.join(cwd, path))

- file_path = os.path.join(path, filename)

- if os.path.exists(os.path.normpath(file_path)):

- return file_path

-

- def load_and_generate_flow_graph(self, file_path, out_path=None, hier_only=False):

- """Loads a flow graph from file and generates it"""

- Messages.set_indent(len(self._auto_hier_block_generate_chain))

- Messages.send('>>> Loading: {}\n'.format(file_path))

- if file_path in self._auto_hier_block_generate_chain:

- Messages.send(' >>> Warning: cyclic hier_block dependency\n')

- return None, None

- self._auto_hier_block_generate_chain.add(file_path)

- try:

- flow_graph = self.get_new_flow_graph()

- flow_graph.grc_file_path = file_path

- # Other, nested hier_blocks might be auto-loaded here

- flow_graph.import_data(self.parse_flow_graph(file_path))

- flow_graph.rewrite()

- flow_graph.validate()

- if not flow_graph.is_valid():

- raise Exception('Flowgraph invalid')

- if hier_only and not flow_graph.get_option('generate_options').startswith('hb'):

- raise Exception('Not a hier block')

- except Exception as e:

- Messages.send('>>> Load Error: {}: {}\n'.format(file_path, str(e)))

- return None, None

- finally:

- self._auto_hier_block_generate_chain.discard(file_path)

- Messages.set_indent(len(self._auto_hier_block_generate_chain))

-

- try:

- generator = self.Generator(flow_graph, out_path or file_path)

- Messages.send('>>> Generating: {}\n'.format(generator.file_path))

- generator.write()

- except Exception as e:

- Messages.send('>>> Generate Error: {}: {}\n'.format(file_path, str(e)))

- return None, None

-

- if flow_graph.get_option('generate_options').startswith('hb'):

- self.load_block_xml(generator.get_file_path_xml())

- return flow_graph, generator.file_path

-

- def build_block_library(self):

- """load the blocks and block tree from the search paths"""

- self._docstring_extractor.start()

- # Reset

- self.blocks.clear()

- self._blocks_n.clear()

- self._block_categories.clear()

- self.domains.clear()

- self.connection_templates.clear()

- ParseXML.xml_failures.clear()

-

- # Try to parse and load blocks

- for xml_file in self.iter_xml_files():

- try:

- if xml_file.endswith("block_tree.xml"):

- self.load_category_tree_xml(xml_file)

- elif xml_file.endswith('domain.xml'):

- self.load_domain_xml(xml_file)

- else:

- self.load_block_xml(xml_file)

- except ParseXML.XMLSyntaxError as e:

- # print >> sys.stderr, 'Warning: Block validation failed:\n\t%s\n\tIgnoring: %s' % (e, xml_file)

- pass

- except Exception as e:

- print >> sys.stderr, 'Warning: XML parsing failed:\n\t%r\n\tIgnoring: %s' % (e, xml_file)

-

- # Add blocks to block tree

- for key, block in self.blocks.iteritems():

- category = self._block_categories.get(key, block.category)

- # Blocks with empty categories are hidden

- if not category:

- continue

- root = category[0]

- if root.startswith('[') and root.endswith(']'):

- category[0] = root[1:-1]

- else:

- category.insert(0, Constants.DEFAULT_BLOCK_MODULE_NAME)

- block.category = category

-

- self._docstring_extractor.finish()

- # self._docstring_extractor.wait()

-

- hide_bokeh_gui_options_if_not_installed(self.blocks['options'])

-

- def iter_xml_files(self):

- """Iterator for block descriptions and category trees"""

- for block_path in self.config.block_paths:

- if os.path.isfile(block_path):

- yield block_path

- elif os.path.isdir(block_path):

- for dirpath, dirnames, filenames in os.walk(block_path):

- for filename in sorted(filter(lambda f: f.endswith('.xml'), filenames)):

- yield os.path.join(dirpath, filename)

-

- def load_block_xml(self, xml_file):

- """Load block description from xml file"""

- # Validate and import

- ParseXML.validate_dtd(xml_file, Constants.BLOCK_DTD)

- n = ParseXML.from_file(xml_file).find('block')

- n['block_wrapper_path'] = xml_file # inject block wrapper path

- # Get block instance and add it to the list of blocks

- block = self.Block(self._flow_graph, n)

- key = block.get_key()

- if key in self.blocks:

- print >> sys.stderr, 'Warning: Block with key "{}" already exists.\n\tIgnoring: {}'.format(key, xml_file)

- else: # Store the block

- self.blocks[key] = block

- self._blocks_n[key] = n

-

- self._docstring_extractor.query(

- block.get_key(),

- block.get_imports(raw=True),

- block.get_make(raw=True)

- )

-

- def load_category_tree_xml(self, xml_file):

- """Validate and parse category tree file and add it to list"""

- ParseXML.validate_dtd(xml_file, Constants.BLOCK_TREE_DTD)

- xml = ParseXML.from_file(xml_file)

- path = []

-

- def load_category(cat_n):

- path.append(cat_n.find('name').strip())

- for block_key in cat_n.findall('block'):

- if block_key not in self._block_categories:

- self._block_categories[block_key] = list(path)

- for sub_cat_n in cat_n.findall('cat'):

- load_category(sub_cat_n)

- path.pop()

-

- load_category(xml.find('cat'))

-

- def load_domain_xml(self, xml_file):

- """Load a domain properties and connection templates from XML"""

- ParseXML.validate_dtd(xml_file, Constants.DOMAIN_DTD)

- n = ParseXML.from_file(xml_file).find('domain')

-

- key = n.find('key')

- if not key:

- print >> sys.stderr, 'Warning: Domain with empty key.\n\tIgnoring: {0}'.format(xml_file)

- return

- if key in self.domains: # test against repeated keys

- print >> sys.stderr, 'Warning: Domain with key "{}" already exists.\n\tIgnoring: {}'.format(key, xml_file)

- return

-

- #to_bool = lambda s, d: d if s is None else s.lower() not in ('false', 'off', '0', '')

- def to_bool(s, d):

- if s is not None:

- return s.lower() not in ('false', 'off', '0', '')

- return d

-

- color = n.find('color') or ''

- try:

- import gtk # ugly but handy

- gtk.gdk.color_parse(color)

- except (ValueError, ImportError):

- if color: # no color is okay, default set in GUI

- print >> sys.stderr, 'Warning: Can\'t parse color code "{}" for domain "{}" '.format(color, key)

- color = None

-

- self.domains[key] = dict(

- name=n.find('name') or key,

- multiple_sinks=to_bool(n.find('multiple_sinks'), True),

- multiple_sources=to_bool(n.find('multiple_sources'), False),

- color=color

- )

- for connection_n in n.findall('connection'):

- key = (connection_n.find('source_domain'), connection_n.find('sink_domain'))

- if not all(key):

- print >> sys.stderr, 'Warning: Empty domain key(s) in connection template.\n\t{}'.format(xml_file)

- elif key in self.connection_templates:

- print >> sys.stderr, 'Warning: Connection template "{}" already exists.\n\t{}'.format(key, xml_file)

- else:

- self.connection_templates[key] = connection_n.find('make') or ''

-

- def _save_docstring_extraction_result(self, key, docstrings):

- docs = {}

- for match, docstring in docstrings.iteritems():

- if not docstring or match.endswith('_sptr'):

- continue

- docstring = docstring.replace('\n\n', '\n').strip()

- docs[match] = docstring

- self.block_docstrings[key] = docs

-

- ##############################################

- # Access

- ##############################################

-

- def parse_flow_graph(self, flow_graph_file):

- """

- Parse a saved flow graph file.

- Ensure that the file exists, and passes the dtd check.

-

- Args:

- flow_graph_file: the flow graph file

-

- Returns:

- nested data

- @throws exception if the validation fails

- """

- flow_graph_file = flow_graph_file or self.config.default_flow_graph

- open(flow_graph_file, 'r').close() # Test open

- ParseXML.validate_dtd(flow_graph_file, Constants.FLOW_GRAPH_DTD)

- return ParseXML.from_file(flow_graph_file)

-

- def get_new_flow_graph(self):

- return self.FlowGraph(platform=self)

-

- def get_blocks(self):

- return self.blocks.values()

-

- def get_new_block(self, flow_graph, key):

- return self.Block(flow_graph, n=self._blocks_n[key])

-

- def get_colors(self):

- return [(name, color) for name, key, sizeof, color in Constants.CORE_TYPES]

-"""

-Copyright 2008-2017 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

-"""

-

-from itertools import chain

-

-from .Constants import DEFAULT_DOMAIN, GR_STREAM_DOMAIN, GR_MESSAGE_DOMAIN

-from .Element import Element

-

-from . import Constants

-

-

-class LoopError(Exception):

- pass

-

-

-def _upstream_ports(port):

- if port.is_sink:

- return _sources_from_virtual_sink_port(port)

- else:

- return _sources_from_virtual_source_port(port)

-

-

-def _sources_from_virtual_sink_port(sink_port, _traversed=None):

- """

- Resolve the source port that is connected to the given virtual sink port.

- Use the get source from virtual source to recursively resolve subsequent ports.

- """

- source_ports_per_virtual_connection = (

- # there can be multiple ports per virtual connection

- _sources_from_virtual_source_port(c.get_source(), _traversed) # type: list

- for c in sink_port.get_enabled_connections()

- )

- return list(chain(*source_ports_per_virtual_connection)) # concatenate generated lists of ports

-

-

-def _sources_from_virtual_source_port(source_port, _traversed=None):

- """

- Recursively resolve source ports over the virtual connections.

- Keep track of traversed sources to avoid recursive loops.

- """

- _traversed = set(_traversed or []) # a new set!

- if source_port in _traversed:

- raise LoopError('Loop found when resolving port type')

- _traversed.add(source_port)

-

- block = source_port.get_parent()

- flow_graph = block.get_parent()

-

- if not block.is_virtual_source():

- return [source_port] # nothing to resolve, we're done

-

- stream_id = block.get_param('stream_id').get_value()

-

- # currently the validation does not allow multiple virtual sinks and one virtual source

- # but in the future it may...

- connected_virtual_sink_blocks = (

- b for b in flow_graph.get_enabled_blocks()

- if b.is_virtual_sink() and b.get_param('stream_id').get_value() == stream_id

- )

- source_ports_per_virtual_connection = (

- _sources_from_virtual_sink_port(b.get_sinks()[0], _traversed) # type: list

- for b in connected_virtual_sink_blocks

- )

- return list(chain(*source_ports_per_virtual_connection)) # concatenate generated lists of ports

-

-

-def _downstream_ports(port):

- if port.is_source:

- return _sinks_from_virtual_source_port(port)

- else:

- return _sinks_from_virtual_sink_port(port)

-

-

-def _sinks_from_virtual_source_port(source_port, _traversed=None):

- """

- Resolve the sink port that is connected to the given virtual source port.

- Use the get sink from virtual sink to recursively resolve subsequent ports.

- """

- sink_ports_per_virtual_connection = (

- # there can be multiple ports per virtual connection

- _sinks_from_virtual_sink_port(c.get_sink(), _traversed) # type: list

- for c in source_port.get_enabled_connections()

- )

- return list(chain(*sink_ports_per_virtual_connection)) # concatenate generated lists of ports

-

-

-def _sinks_from_virtual_sink_port(sink_port, _traversed=None):

- """

- Recursively resolve sink ports over the virtual connections.

- Keep track of traversed sinks to avoid recursive loops.

- """

- _traversed = set(_traversed or []) # a new set!

- if sink_port in _traversed:

- raise LoopError('Loop found when resolving port type')

- _traversed.add(sink_port)

-

- block = sink_port.get_parent()

- flow_graph = block.get_parent()

-

- if not block.is_virtual_sink():

- return [sink_port]

-

- stream_id = block.get_param('stream_id').get_value()

-

- connected_virtual_source_blocks = (

- b for b in flow_graph.get_enabled_blocks()

- if b.is_virtual_source() and b.get_param('stream_id').get_value() == stream_id

- )

- sink_ports_per_virtual_connection = (

- _sinks_from_virtual_source_port(b.get_sources()[0], _traversed) # type: list

- for b in connected_virtual_source_blocks

- )

- return list(chain(*sink_ports_per_virtual_connection)) # concatenate generated lists of ports

-

-

-class Port(Element):

-

- is_port = True

-

- def __init__(self, block, n, dir):

- """

- Make a new port from nested data.

-

- Args:

- block: the parent element

- n: the nested odict

- dir: the direction

- """

- self._n = n

- if n['type'] == 'message':

- n['domain'] = GR_MESSAGE_DOMAIN

- if 'domain' not in n:

- n['domain'] = DEFAULT_DOMAIN

- elif n['domain'] == GR_MESSAGE_DOMAIN:

- n['key'] = n['name']

- n['type'] = 'message' # For port color

- if not n.find('key'):

- n['key'] = str(next(block.port_counters[dir == 'source']))

-

- # Build the port

- Element.__init__(self, block)

- # Grab the data

- self._name = n['name']

- self._key = n['key']

- self._type = n['type'] or ''

- self._domain = n['domain']

- self._hide = n.find('hide') or ''

- self._dir = dir

- self._hide_evaluated = False # Updated on rewrite()

-

- self._nports = n.find('nports') or ''

- self._vlen = n.find('vlen') or ''

- self._optional = n.find('optional') or ''

- self._optional_evaluated = False # Updated on rewrite()

- self._clones = [] # References to cloned ports (for nports > 1)

-

- def __str__(self):

- if self.is_source:

- return 'Source - {}({})'.format(self.get_name(), self.get_key())

- if self.is_sink:

- return 'Sink - {}({})'.format(self.get_name(), self.get_key())

-

- def get_types(self):

- return Constants.TYPE_TO_SIZEOF.keys()

-

- def is_type_empty(self):

- return not self._n['type'] or not self.get_parent().resolve_dependencies(self._n['type'])

-

- def validate(self):

- if self.get_type() not in self.get_types():

- self.add_error_message('Type "{}" is not a possible type.'.format(self.get_type()))

- platform = self.get_parent().get_parent().get_parent()

- if self.get_domain() not in platform.domains:

- self.add_error_message('Domain key "{}" is not registered.'.format(self.get_domain()))

- if not self.get_enabled_connections() and not self.get_optional():

- self.add_error_message('Port is not connected.')

-

- def rewrite(self):

- """

- Handle the port cloning for virtual blocks.

- """

- del self._error_messages[:]

- if self.is_type_empty():

- self.resolve_empty_type()

-

- hide = self.get_parent().resolve_dependencies(self._hide).strip().lower()

- self._hide_evaluated = False if hide in ('false', 'off', '0') else bool(hide)

- optional = self.get_parent().resolve_dependencies(self._optional).strip().lower()

- self._optional_evaluated = False if optional in ('false', 'off', '0') else bool(optional)

-

- # Update domain if was deduced from (dynamic) port type

- type_ = self.get_type()

- if self._domain == GR_STREAM_DOMAIN and type_ == "message":

- self._domain = GR_MESSAGE_DOMAIN

- self._key = self._name

- if self._domain == GR_MESSAGE_DOMAIN and type_ != "message":

- self._domain = GR_STREAM_DOMAIN

- self._key = '0' # Is rectified in rewrite()

-

- def resolve_virtual_source(self):

- """Only used by Generator after validation is passed"""

- return _upstream_ports(self)

-

- def resolve_empty_type(self):

- def find_port(finder):

- try:

- return next((p for p in finder(self) if not p.is_type_empty()), None)

- except LoopError as error:

- self.add_error_message(str(error))

- except (StopIteration, Exception) as error:

- pass

-

- try:

- port = find_port(_upstream_ports) or find_port(_downstream_ports)

- self._type = str(port.get_type())

- self._vlen = str(port.get_vlen())

- except Exception:

- # Reset type and vlen

- self._type = self._vlen = ''

-

- def get_vlen(self):

- """

- Get the vector length.

- If the evaluation of vlen cannot be cast to an integer, return 1.

-

- Returns:

- the vector length or 1

- """

- vlen = self.get_parent().resolve_dependencies(self._vlen)

- try:

- return int(self.get_parent().get_parent().evaluate(vlen))

- except:

- return 1

-

- def get_nports(self):

- """

- Get the number of ports.

- If already blank, return a blank

- If the evaluation of nports cannot be cast to a positive integer, return 1.

-

- Returns:

- the number of ports or 1

- """

- if self._nports == '':

- return ''

-

- nports = self.get_parent().resolve_dependencies(self._nports)

- try:

- return max(1, int(self.get_parent().get_parent().evaluate(nports)))

- except:

- return 1

-

- def get_optional(self):

- return self._optional_evaluated

-

- def get_color(self):

- """

- Get the color that represents this port's type.

- Codes differ for ports where the vec length is 1 or greater than 1.

-

- Returns:

- a hex color code.

- """

- try:

- color = Constants.TYPE_TO_COLOR[self.get_type()]

- vlen = self.get_vlen()

- if vlen == 1:

- return color

- color_val = int(color[1:], 16)

- r = (color_val >> 16) & 0xff

- g = (color_val >> 8) & 0xff

- b = (color_val >> 0) & 0xff

- dark = (0, 0, 30, 50, 70)[min(4, vlen)]

- r = max(r-dark, 0)

- g = max(g-dark, 0)

- b = max(b-dark, 0)

- # TODO: Change this to .format()

- return '#%.2x%.2x%.2x' % (r, g, b)

- except:

- return '#FFFFFF'

-

- def get_clones(self):

- """

- Get the clones of this master port (nports > 1)

-

- Returns:

- a list of ports

- """

- return self._clones

-

- def add_clone(self):

- """

- Create a clone of this (master) port and store a reference in self._clones.

-

- The new port name (and key for message ports) will have index 1... appended.

- If this is the first clone, this (master) port will get a 0 appended to its name (and key)

-

- Returns:

- the cloned port

- """

- # Add index to master port name if there are no clones yet

- if not self._clones:

- self._name = self._n['name'] + '0'

- # Also update key for none stream ports

- if not self._key.isdigit():

- self._key = self._name

-

- # Prepare a copy of the odict for the clone

- n = self._n.copy()

- # Remove nports from the key so the copy cannot be a duplicator

- if 'nports' in n:

- n.pop('nports')

- n['name'] = self._n['name'] + str(len(self._clones) + 1)

- # Dummy value 99999 will be fixed later

- n['key'] = '99999' if self._key.isdigit() else n['name']

-

- # Clone

- port = self.__class__(self.get_parent(), n, self._dir)

- self._clones.append(port)

- return port

-

- def remove_clone(self, port):

- """

- Remove a cloned port (from the list of clones only)

- Remove the index 0 of the master port name (and key9 if there are no more clones left

- """

- self._clones.remove(port)

- # Remove index from master port name if there are no more clones

- if not self._clones:

- self._name = self._n['name']

- # Also update key for none stream ports

- if not self._key.isdigit():

- self._key = self._name

-

- def get_name(self):

- number = ''

- if self.get_type() == 'bus':

- busses = filter(lambda a: a._dir == self._dir, self.get_parent().get_ports_gui())

- number = str(busses.index(self)) + '#' + str(len(self.get_associated_ports()))

- return self._name + number

-

- def get_key(self):

- return self._key

-

- @property

- def is_sink(self):

- return self._dir == 'sink'

-

- @property

- def is_source(self):

- return self._dir == 'source'

-

- def get_type(self):

- return self.get_parent().resolve_dependencies(self._type)

-

- def get_domain(self):

- return self._domain

-

- def get_hide(self):

- return self._hide_evaluated

-

- def get_connections(self):

- """

- Get all connections that use this port.

-

- Returns:

- a list of connection objects

- """

- connections = self.get_parent().get_parent().connections

- connections = filter(lambda c: c.get_source() is self or c.get_sink() is self, connections)

- return connections

-

- def get_enabled_connections(self):

- """

- Get all enabled connections that use this port.

-

- Returns:

- a list of connection objects

- """

- return filter(lambda c: c.get_enabled(), self.get_connections())

-

- def get_associated_ports(self):

- if not self.get_type() == 'bus':

- return [self]

- else:

- if self.is_source:

- get_ports = self.get_parent().get_sources

- bus_structure = self.get_parent().current_bus_structure['source']

- else:

- get_ports = self.get_parent().get_sinks

- bus_structure = self.get_parent().current_bus_structure['sink']

-

- ports = [i for i in get_ports() if not i.get_type() == 'bus']

- if bus_structure:

- busses = [i for i in get_ports() if i.get_type() == 'bus']

- bus_index = busses.index(self)

- ports = filter(lambda a: ports.index(a) in bus_structure[bus_index], ports)

- return ports

-<!--

-Copyright 2008 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

--->

-<!--

- gnuradio_python.blocks.dtd

- Josh Blum

- The document type definition for blocks.

- -->

-<!--

- Top level element.

- A block contains a name, ...parameters list, and list of IO ports.

- -->

-<!ELEMENT block (name, key, category?, throttle?, flags?, import*, var_make?, var_value?,

- make, callback*, param_tab_order?, param*, bus_sink?, bus_source?, check*,

- sink*, source*, bus_structure_sink?, bus_structure_source?, doc?, grc_source?)>

-<!--

- Sub level elements.

- -->

-<!ELEMENT param_tab_order (tab+)>

-<!ELEMENT param (base_key?, name, key, value?, type?, hide?, option*, tab?)>

-<!ELEMENT option (name, key, opt*)>

-<!ELEMENT sink (name, type, vlen?, domain?, nports?, optional?, hide?)>

-<!ELEMENT source (name, type, vlen?, domain?, nports?, optional?, hide?)>

-<!--

- Bottom level elements.

- Character data only.

- -->

-<!ELEMENT category (#PCDATA)>

-<!ELEMENT import (#PCDATA)>

-<!ELEMENT doc (#PCDATA)>

-<!ELEMENT grc_source (#PCDATA)>

-<!ELEMENT tab (#PCDATA)>

-<!ELEMENT name (#PCDATA)>

-<!ELEMENT base_key (#PCDATA)>

-<!ELEMENT key (#PCDATA)>

-<!ELEMENT check (#PCDATA)>

-<!ELEMENT bus_sink (#PCDATA)>

-<!ELEMENT bus_source (#PCDATA)>

-<!ELEMENT opt (#PCDATA)>

-<!ELEMENT type (#PCDATA)>

-<!ELEMENT hide (#PCDATA)>

-<!ELEMENT vlen (#PCDATA)>

-<!ELEMENT domain (#PCDATA)>

-<!ELEMENT nports (#PCDATA)>

-<!ELEMENT bus_structure_sink (#PCDATA)>

-<!ELEMENT bus_structure_source (#PCDATA)>

-<!ELEMENT var_make (#PCDATA)>

-<!ELEMENT var_value (#PCDATA)>

-<!ELEMENT make (#PCDATA)>

-<!ELEMENT value (#PCDATA)>

-<!ELEMENT callback (#PCDATA)>

-<!ELEMENT optional (#PCDATA)>

-<!ELEMENT throttle (#PCDATA)>

-<!ELEMENT flags (#PCDATA)>

-# Copyright 2008, 2009, 2015, 2016 Free Software Foundation, Inc.

-from . import Platform, FlowGraph, Block

-def lazy_property(func):

- return func

-class Element(object):

- def __init__(self, parent=None):

- ...

- @property

- def parent(self):

- ...

- def get_parent_by_type(self, cls):

- parent = self.parent

- if parent is None:

- return None

- elif isinstance(parent, cls):

- return parent

- else:

- return parent.get_parent_by_type(cls)

-

- @lazy_property

- def parent_platform(self): -> Platform.Platform

- ...

-

- @lazy_property

- def parent_flowgraph(self): -> FlowGraph.FlowGraph

- ...

-

- @lazy_property

- def parent_block(self): -> Block.Block

- ...

-<?xml version="1.0"?>

-<!--

-##Default Flow Graph:

-## include an options block and a variable for sample rate

- -->

-<flow_graph>

- <block>

- <key>options</key>

- <param>

- <key>id</key>

- <value>top_block</value>

- </param>

- <param>

- <key>_coordinate</key>

- <value>(8, 8)</value>

- </param>

- <param>

- <key>_rotation</key>

- <value>0</value>

- </param>

- </block>

- <block>

- <key>variable</key>

- <param>

- <key>id</key>

- <value>samp_rate</value>

- </param>

- <param>

- <key>value</key>

- <value>32000</value>

- </param>

- <param>

- <key>_coordinate</key>

- <value>(8, 160)</value>

- </param>

- <param>

- <key>_rotation</key>

- <value>0</value>

- </param>

- </block>

-</flow_graph>

-<!--

-Copyright 2014 Free Software Foundation, Inc.

-This file is part of GNU Radio

-

-GNU Radio Companion is free software; you can redistribute it and/or

-modify it under the terms of the GNU General Public License

-as published by the Free Software Foundation; either version 2

-of the License, or (at your option) any later version.

-

-GNU Radio Companion is distributed in the hope that it will be useful,

-but WITHOUT ANY WARRANTY; without even the implied warranty of

-MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-GNU General Public License for more details.

-

-You should have received a copy of the GNU General Public License

-along with this program; if not, write to the Free Software

-Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

--->

-<!ELEMENT domain (name, key, color?, multiple_sinks?, multiple_sources?, connection*)>

-<!--

- Sub level elements.

- -->

-<!ELEMENT connection (source_domain, sink_domain, make)>

-<!--

- Bottom level elements.

- Character data only.

- -->

-<!ELEMENT name (#PCDATA)>

-<!ELEMENT key (#PCDATA)>

-<!ELEMENT multiple_sinks (#PCDATA)>

-<!ELEMENT multiple_sources (#PCDATA)>

-<!ELEMENT color (#PCDATA)>

-<!ELEMENT make (#PCDATA)>

-<!ELEMENT source_domain (#PCDATA)>

-<!ELEMENT sink_domain (#PCDATA)>

-# Copyright 2011 Free Software Foundation, Inc.

-#

-# This file is part of GNU Radio

-#

-# GNU Radio is free software; you can redistribute it and/or modify

-# it under the terms of the GNU General Public License as published by

-# the Free Software Foundation; either version 3, or (at your option)

-# any later version.

-#

-# GNU Radio is distributed in the hope that it will be useful,

-# but WITHOUT ANY WARRANTY; without even the implied warranty of

-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

-# GNU General Public License for more details.

-#

-# You should have received a copy of the GNU General Public License

-# along with GNU Radio; see the file COPYING. If not, write to

-# the Free Software Foundation, Inc., 51 Franklin Street,

-# Boston, MA 02110-1301, USA.

-

-file(GLOB py_files "*.py")

-

-GR_PYTHON_INSTALL(

- FILES ${py_files}

- DESTINATION ${GR_PYTHON_DIR}/gnuradio/grc/core/generator

-)

-

-install(FILES

- flow_graph.tmpl

- DESTINATION ${GR_PYTHON_DIR}/gnuradio/grc/core/generator

-)

-class FlowGraphProxy(object):

- self._fg = fg

- return getattr(self._fg, item)

- return filter(lambda p: p['type'] != "message", self.get_hier_block_io(direction))

- return filter(lambda p: p['type'] == "message", self.get_hier_block_io(direction))

- 'label': str(pad.get_param('label').get_evaluated()),

- 'type': str(pad.get_param('type').get_evaluated()),

- 'vlen': str(pad.get_param('vlen').get_value()),

- 'size': pad.get_param('type').get_opt('size'),

- 'optional': bool(pad.get_param('optional').get_evaluated()),

- num_ports = pad.get_param('num_streams').get_evaluated()

- for i in xrange(num_ports):

- pads = filter(lambda b: b.get_key() == 'pad_source', self.get_enabled_blocks())

- return sorted(pads, lambda x, y: cmp(x.get_id(), y.get_id()))

- pads = filter(lambda b: b.get_key() == 'pad_sink', self.get_enabled_blocks())

- return sorted(pads, lambda x, y: cmp(x.get_id(), y.get_id()))

- is_message_pad = pad.get_param('type').get_evaluated() == "message"

- if port.get_parent() == pad:

- key = pad.get_param('label').get_value()

- key = str(key_offset + int(port.get_key()))

- key_offset += len(pad.get_ports())

- return -1 \ No newline at end of file

-import codecs

-import tempfile

-from Cheetah.Template import Template

-from .FlowGraphProxy import FlowGraphProxy

-from .. import ParseXML, Messages

-from ..Constants import (

- TOP_BLOCK_FILE_MODE, BLOCK_FLAG_NEED_QT_GUI,

- HIER_BLOCK_FILE_MODE, BLOCK_DTD

-)

-from ..utils import expr_utils, odict

-FLOW_GRAPH_TEMPLATE = os.path.join(DATA_DIR, 'flow_graph.tmpl')

-

-

-class TopBlockGenerator(object):

-

- def __init__(self, flow_graph, file_path):

- """

- Initialize the top block generator object.

-

- Args:

- flow_graph: the flow graph object

- file_path: the path to write the file to

- """

- self._flow_graph = FlowGraphProxy(flow_graph)

- self._generate_options = self._flow_graph.get_option('generate_options')

- self._mode = TOP_BLOCK_FILE_MODE

- dirname = os.path.dirname(file_path)

- # Handle the case where the directory is read-only

- # In this case, use the system's temp directory

- if not os.access(dirname, os.W_OK):

- dirname = tempfile.gettempdir()

- filename = self._flow_graph.get_option('id') + '.py'

- self.file_path = os.path.join(dirname, filename)

- self._dirname = dirname

-

- def get_file_path(self):

- return self.file_path

-

- def write(self):

- """generate output and write it to files"""

- # Do throttle warning

- throttling_blocks = filter(lambda b: b.throtteling(), self._flow_graph.get_enabled_blocks())

- if not throttling_blocks and not self._generate_options.startswith('hb'):

- Messages.send_warning("This flow graph may not have flow control: "

- "no audio or RF hardware blocks found. "

- "Add a Misc->Throttle block to your flow "

- "graph to avoid CPU congestion.")

- if len(throttling_blocks) > 1:

- keys = set(map(lambda b: b.get_key(), throttling_blocks))

- if len(keys) > 1 and 'blocks_throttle' in keys:

- Messages.send_warning("This flow graph contains a throttle "

- "block and another rate limiting block, "

- "e.g. a hardware source or sink. "

- "This is usually undesired. Consider "

- "removing the throttle block.")

- # Generate

- for filename, data in self._build_python_code_from_template():

- with codecs.open(filename, 'w', encoding='utf-8') as fp:

- fp.write(data)

- if filename == self.file_path:

- try:

- os.chmod(filename, self._mode)

- except:

- pass

-

- def _build_python_code_from_template(self):

- """

- Convert the flow graph to python code.

-

- Returns:

- a string of python code

- """

- output = list()

-

- fg = self._flow_graph

- title = fg.get_option('title') or fg.get_option('id').replace('_', ' ').title()

- imports = fg.get_imports()

- variables = fg.get_variables()

- parameters = fg.get_parameters()

- monitors = fg.get_monitors()

-

- # List of blocks not including variables and imports and parameters and disabled

- def _get_block_sort_text(block):

- code = block.get_make().replace(block.get_id(), ' ')

- try:

- code += block.get_param('gui_hint').get_value() # Newer gui markup w/ qtgui

- except:

- pass

- return code

-

- blocks_all = expr_utils.sort_objects(

- filter(lambda b: b.get_enabled() and not b.get_bypassed(), fg.blocks),

- lambda b: b.get_id(), _get_block_sort_text

- )

- deprecated_block_keys = set(block.get_name() for block in blocks_all if block.is_deprecated)

- for key in deprecated_block_keys:

- Messages.send_warning("The block {!r} is deprecated.".format(key))

-

- # List of regular blocks (all blocks minus the special ones)

- blocks = filter(lambda b: b not in (imports + parameters), blocks_all)

-

- for block in blocks:

- key = block.get_key()

- file_path = os.path.join(self._dirname, block.get_id() + '.py')

- if key == 'epy_block':

- src = block.get_param('_source_code').get_value()

- output.append((file_path, src))

- elif key == 'epy_module':

- src = block.get_param('source_code').get_value()

- output.append((file_path, src))

-

- # Filter out virtual sink connections

- def cf(c):

- return not (c.is_bus() or c.get_sink().get_parent().is_virtual_sink())

- connections = filter(cf, fg.get_enabled_connections())

-

- # Get the virtual blocks and resolve their connections

- virtual = filter(lambda c: c.get_source().get_parent().is_virtual_source(), connections)

- for connection in virtual:

- sink = connection.get_sink()

- for source in connection.get_source().resolve_virtual_source():

- resolved = fg.get_parent().Connection(flow_graph=fg, porta=source, portb=sink)

- connections.append(resolved)

- # Remove the virtual connection

- connections.remove(connection)

-

- # Bypassing blocks: Need to find all the enabled connections for the block using

- # the *connections* object rather than get_connections(). Create new connections

- # that bypass the selected block and remove the existing ones. This allows adjacent

- # bypassed blocks to see the newly created connections to downstream blocks,

- # allowing them to correctly construct bypass connections.

- bypassed_blocks = fg.get_bypassed_blocks()

- for block in bypassed_blocks:

- # Get the upstream connection (off of the sink ports)

- # Use *connections* not get_connections()

- source_connection = filter(lambda c: c.get_sink() == block.get_sinks()[0], connections)

- # The source connection should never have more than one element.

- assert (len(source_connection) == 1)

-

- # Get the source of the connection.

- source_port = source_connection[0].get_source()

-

- # Loop through all the downstream connections

- for sink in filter(lambda c: c.get_source() == block.get_sources()[0], connections):

- if not sink.get_enabled():

- # Ignore disabled connections

- continue

- sink_port = sink.get_sink()

- connection = fg.get_parent().Connection(flow_graph=fg, porta=source_port, portb=sink_port)

- connections.append(connection)

- # Remove this sink connection

- connections.remove(sink)

- # Remove the source connection

- connections.remove(source_connection[0])

-

- # List of connections where each endpoint is enabled (sorted by domains, block names)

- connections.sort(key=lambda c: (

- c.get_source().get_domain(), c.get_sink().get_domain(),

- c.get_source().get_parent().get_id(), c.get_sink().get_parent().get_id()

- ))

-

- connection_templates = fg.get_parent().connection_templates

-

- # List of variable names

- var_ids = [var.get_id() for var in parameters + variables]

- replace_dict = dict((var_id, 'self.' + var_id) for var_id in var_ids)

- callbacks_all = []

- for block in blocks_all:

- callbacks_all.extend(expr_utils.expr_replace(cb, replace_dict) for cb in block.get_callbacks())

-

- # Map var id to callbacks

- def uses_var_id():

- used = expr_utils.get_variable_dependencies(callback, [var_id])

- return used and 'self.' + var_id in callback # callback might contain var_id itself

-

- callbacks = {}

- for var_id in var_ids:

- callbacks[var_id] = [callback for callback in callbacks_all if uses_var_id()]

-

- # Load the namespace

- namespace = {

- 'title': title,

- 'imports': imports,

- 'flow_graph': fg,

- 'variables': variables,

- 'parameters': parameters,

- 'monitors': monitors,

- 'blocks': blocks,

- 'connections': connections,

- 'connection_templates': connection_templates,

- 'generate_options': self._generate_options,

- 'callbacks': callbacks,

- }

- # Build the template

- t = Template(open(FLOW_GRAPH_TEMPLATE, 'r').read(), namespace)

- output.append((self.file_path, "\n".join(line.rstrip() for line in str(t).split("\n"))))

- return output

-

-

-class HierBlockGenerator(TopBlockGenerator):

- """Extends the top block generator to also generate a block XML 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 xml goes into HIER_BLOCK_LIB_DIR)

- """

- TopBlockGenerator.__init__(self, flow_graph, file_path)

- platform = flow_graph.get_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 = HIER_BLOCK_FILE_MODE

- self.file_path = os.path.join(hier_block_lib_dir, self._flow_graph.get_option('id') + '.py')

- self._file_path_xml = self.file_path + '.xml'

-

- def get_file_path_xml(self):

- return self._file_path_xml

-

- def write(self):

- """generate output and write it to files"""

- TopBlockGenerator.write(self)

- ParseXML.to_file(self._build_block_n_from_flow_graph_io(), self.get_file_path_xml())

- ParseXML.validate_dtd(self.get_file_path_xml(), BLOCK_DTD)

- try:

- os.chmod(self.get_file_path_xml(), self._mode)

- except:

- pass

-

- def _build_block_n_from_flow_graph_io(self):

- """

- Generate a block XML nested data from the flow graph IO

-

- Returns:

- a xml node tree

- """

- # Extract info from the flow graph

- block_key = self._flow_graph.get_option('id')

- parameters = self._flow_graph.get_parameters()

-

- def var_or_value(name):

- if name in map(lambda p: p.get_id(), parameters):

- return "$"+name

- return name

-

- # Build the nested data

- block_n = odict()

- block_n['name'] = self._flow_graph.get_option('title') or \

- self._flow_graph.get_option('id').replace('_', ' ').title()

- block_n['key'] = block_key

- block_n['category'] = self._flow_graph.get_option('category')

- block_n['import'] = "from {0} import {0} # grc-generated hier_block".format(

- self._flow_graph.get_option('id'))

- # Make data

- if parameters:

- block_n['make'] = '{cls}(\n {kwargs},\n)'.format(

- cls=block_key,

- kwargs=',\n '.join(

- '{key}=${key}'.format(key=param.get_id()) for param in parameters

- ),

- )

- else:

- block_n['make'] = '{cls}()'.format(cls=block_key)

- # Callback data

- block_n['callback'] = [

- 'set_{key}(${key})'.format(key=param.get_id()) for param in parameters

- ]

-

- # Parameters

- block_n['param'] = list()

- for param in parameters:

- param_n = odict()

- param_n['name'] = param.get_param('label').get_value() or param.get_id()

- param_n['key'] = param.get_id()

- param_n['value'] = param.get_param('value').get_value()

- param_n['type'] = 'raw'

- param_n['hide'] = param.get_param('hide').get_value()

- block_n['param'].append(param_n)

-

- # Bus stuff

- if self._flow_graph.get_bussink():

- block_n['bus_sink'] = '1'

- if self._flow_graph.get_bussrc():

- block_n['bus_source'] = '1'

-

- # Sink/source ports

- for direction in ('sink', 'source'):

- block_n[direction] = list()

- for port in self._flow_graph.get_hier_block_io(direction):

- port_n = odict()

- port_n['name'] = port['label']

- port_n['type'] = port['type']

- if port['type'] != "message":

- port_n['vlen'] = var_or_value(port['vlen'])

- if port['optional']:

- port_n['optional'] = '1'

- block_n[direction].append(port_n)

-

- # More bus stuff

- bus_struct_sink = self._flow_graph.get_bus_structure_sink()

- if bus_struct_sink:

- block_n['bus_structure_sink'] = bus_struct_sink[0].get_param('struct').get_value()

- bus_struct_src = self._flow_graph.get_bus_structure_src()

- if bus_struct_src:

- block_n['bus_structure_source'] = bus_struct_src[0].get_param('struct').get_value()

-

- # Documentation

- block_n['doc'] = "\n".join(field for field in (

- self._flow_graph.get_option('author'),

- self._flow_graph.get_option('description'),

- self.file_path

- ) if field)

- block_n['grc_source'] = str(self._flow_graph.grc_file_path)

-

- n = {'block': block_n}

- return n

-

-

-class QtHierBlockGenerator(HierBlockGenerator):

-

- def _build_block_n_from_flow_graph_io(self):

- n = HierBlockGenerator._build_block_n_from_flow_graph_io(self)

- block_n = n['block']

-

- if not block_n['name'].upper().startswith('QT GUI'):

- block_n['name'] = 'QT GUI ' + block_n['name']

-

- block_n.insert_after('category', 'flags', BLOCK_FLAG_NEED_QT_GUI)

-

- gui_hint_param = odict()

- gui_hint_param['name'] = 'GUI Hint'

- gui_hint_param['key'] = 'gui_hint'

- gui_hint_param['value'] = ''

- gui_hint_param['type'] = 'gui_hint'

- gui_hint_param['hide'] = 'part'

- block_n['param'].append(gui_hint_param)

-

- block_n['make'] += (

- "\n#set $win = 'self.%s' % $id"

- "\n${gui_hint()($win)}"

- )

- return n

-from Generator import Generator

-#if not $generate_options.startswith('hb')

-#!/usr/bin/env python2

-#end if

-# -*- coding: utf-8 -*-

-########################################################

-##Cheetah template - gnuradio_python

-##

-##@param imports the import statements

-##@param flow_graph the flow_graph

-##@param variables the variable blocks

-##@param parameters the parameter blocks

-##@param blocks the signal blocks

-##@param connections the connections

-##@param generate_options the type of flow graph

-##@param callbacks variable id map to callback strings

-########################################################

-#def indent($code)

-#set $code = '\n '.join(str($code).splitlines())

-$code#slurp

-#end def

-#import time

-#set $DIVIDER = '#'*50

-$DIVIDER

-# GNU Radio Python Flow Graph

-# Title: $title

-#if $flow_graph.get_option('author')

-# Author: $flow_graph.get_option('author')

-#end if

-#if $flow_graph.get_option('description')

-# Description: $flow_graph.get_option('description')

-#end if

-# Generated: $time.ctime()

-$DIVIDER

-#if $flow_graph.get_option('thread_safe_setters')

-import threading

-#end if

-

-#if $generate_options == 'qt_gui'

-from distutils.version import StrictVersion

-#end if

-

-## Call XInitThreads as the _very_ first thing.

-## After some Qt import, it's too late

-#if $generate_options == 'qt_gui'

-if __name__ == '__main__':

- import ctypes

- import sys

- if sys.platform.startswith('linux'):

- try:

- x11 = ctypes.cdll.LoadLibrary('libX11.so')

- x11.XInitThreads()

- except:

- print "Warning: failed to XInitThreads()"

-

-#end if

-#

-########################################################

-##Create Imports

-########################################################

-#if $flow_graph.get_option('qt_qss_theme')

-#set imports = $sorted(set($imports + ["import os", "import sys"]))

-#end if

-#if any(imp.endswith("# grc-generated hier_block") for imp in $imports)

-import os

-import sys

-#set imports = $filter(lambda i: i not in ("import os", "import sys"), $imports)

-sys.path.append(os.environ.get('GRC_HIER_PATH', os.path.expanduser('~/.grc_gnuradio')))

-

-#end if

-#for $imp in $imports

-##$(imp.replace(" # grc-generated hier_block", ""))

-$imp

-#end for

-########################################################

-##Create Class

-## Write the class declaration for a top or hier block.

-## The parameter names are the arguments to __init__.

-## Setup the IO signature (hier block only).

-########################################################

-#set $class_name = $flow_graph.get_option('id')

-#set $param_str = ', '.join(['self'] + ['%s=%s'%(param.get_id(), param.get_make()) for param in $parameters])

-#if $generate_options == 'qt_gui'

-from gnuradio import qtgui

-

-class $(class_name)(gr.top_block, Qt.QWidget):

-

- def __init__($param_str):

- gr.top_block.__init__(self, "$title")

- Qt.QWidget.__init__(self)

- self.setWindowTitle("$title")

- qtgui.util.check_set_qss()

- try:

- self.setWindowIcon(Qt.QIcon.fromTheme('gnuradio-grc'))

- except:

- pass

- self.top_scroll_layout = Qt.QVBoxLayout()

- self.setLayout(self.top_scroll_layout)

- self.top_scroll = Qt.QScrollArea()

- self.top_scroll.setFrameStyle(Qt.QFrame.NoFrame)

- self.top_scroll_layout.addWidget(self.top_scroll)

- self.top_scroll.setWidgetResizable(True)

- self.top_widget = Qt.QWidget()

- self.top_scroll.setWidget(self.top_widget)

- self.top_layout = Qt.QVBoxLayout(self.top_widget)

- self.top_grid_layout = Qt.QGridLayout()

- self.top_layout.addLayout(self.top_grid_layout)

-

- self.settings = Qt.QSettings("GNU Radio", "$class_name")

-

- try:

- if StrictVersion(Qt.qVersion()) < StrictVersion("5.0.0"):

- self.restoreGeometry(self.settings.value("geometry").toByteArray())

- else:

- self.restoreGeometry(self.settings.value("geometry"))

- except:

- pass

-#elif $generate_options == 'bokeh_gui'

-

-class $(class_name)(gr.top_block):

- def __init__(self, doc):

- gr.top_block.__init__(self, "$title")

- self.doc = doc

- self.plot_lst = []

- self.widget_lst = []

-#elif $generate_options == 'no_gui'

-

-

-class $(class_name)(gr.top_block):

-

- def __init__($param_str):

- gr.top_block.__init__(self, "$title")

-#elif $generate_options.startswith('hb')

- #set $in_sigs = $flow_graph.get_hier_block_stream_io('in')

- #set $out_sigs = $flow_graph.get_hier_block_stream_io('out')

-

-

-#if $generate_options == 'hb_qt_gui'

-class $(class_name)(gr.hier_block2, Qt.QWidget):

-#else

-class $(class_name)(gr.hier_block2):

-#end if

-#def make_io_sig($io_sigs)

- #set $size_strs = ['%s*%s'%(io_sig['size'], io_sig['vlen']) for io_sig in $io_sigs]

- #if len($io_sigs) == 0

-gr.io_signature(0, 0, 0)#slurp

- #elif len($io_sigs) == 1

-gr.io_signature(1, 1, $size_strs[0])#slurp

- #else

-gr.io_signaturev($(len($io_sigs)), $(len($io_sigs)), [$(', '.join($size_strs))])#slurp

- #end if

-#end def

-

- def __init__($param_str):

- gr.hier_block2.__init__(

- self, "$title",

- $make_io_sig($in_sigs),

- $make_io_sig($out_sigs),

- )

- #for $pad in $flow_graph.get_hier_block_message_io('in')

- self.message_port_register_hier_in("$pad['label']")

- #end for

- #for $pad in $flow_graph.get_hier_block_message_io('out')

- self.message_port_register_hier_out("$pad['label']")

- #end for

- #if $generate_options == 'hb_qt_gui'

-

- Qt.QWidget.__init__(self)

- self.top_layout = Qt.QVBoxLayout()

- self.top_grid_layout = Qt.QGridLayout()

- self.top_layout.addLayout(self.top_grid_layout)

- self.setLayout(self.top_layout)

- #end if

-#end if

-#if $flow_graph.get_option('thread_safe_setters')

-

- self._lock = threading.RLock()

-#end if

-########################################################

-##Create Parameters

-## Set the parameter to a property of self.

-########################################################

-#if $parameters

-

- $DIVIDER

- # Parameters

- $DIVIDER

-#end if

-#for $param in $parameters

- $indent($param.get_var_make())

-#end for

-########################################################

-##Create Variables

-########################################################

-#if $variables

-

- $DIVIDER

- # Variables

- $DIVIDER

-#end if

-#for $var in $variables

- $indent($var.get_var_make())

-#end for

-########################################################

-##Create Blocks

-########################################################

-#if $blocks

-

- $DIVIDER

- # Blocks

- $DIVIDER

-#end if

-#for $blk in filter(lambda b: b.get_make(), $blocks)

- #if $blk in $variables

- $indent($blk.get_make())

- #else

- self.$blk.get_id() = $indent($blk.get_make())

- #if $blk.has_param('alias') and $blk.get_param('alias').get_evaluated()

- (self.$blk.get_id()).set_block_alias("$blk.get_param('alias').get_evaluated()")

- #end if

- #if $blk.has_param('affinity') and $blk.get_param('affinity').get_evaluated()

- (self.$blk.get_id()).set_processor_affinity($blk.get_param('affinity').get_evaluated())

- #end if

- #if (len($blk.get_sources())>0) and $blk.has_param('minoutbuf') and (int($blk.get_param('minoutbuf').get_evaluated()) > 0)

- (self.$blk.get_id()).set_min_output_buffer($blk.get_param('minoutbuf').get_evaluated())

- #end if

- #if (len($blk.get_sources())>0) and $blk.has_param('maxoutbuf') and (int($blk.get_param('maxoutbuf').get_evaluated()) > 0)

- (self.$blk.get_id()).set_max_output_buffer($blk.get_param('maxoutbuf').get_evaluated())

- #end if

- #end if

-#end for

-

-##########################################################

-## Create a layout entry if not manually done for BokehGUI

-##########################################################

-#if $generate_options == 'bokeh_gui'

- if self.widget_lst:

- input_t = bokehgui.BokehLayout.widgetbox(self.widget_lst)

- widgetbox = bokehgui.BokehLayout.WidgetLayout(input_t)

- widgetbox.set_layout(*($flow_graph.get_option('placement')))

- list_obj = [widgetbox] + self.plot_lst

- else:

- list_obj = self.plot_lst

- layout_t = bokehgui.BokehLayout.create_layout(list_obj, "$flow_graph.get_option('sizing_mode')")

- self.doc.add_root(layout_t)

-#end if

-

-########################################################

-##Create Connections

-## The port name should be the id of the parent block.

-## However, port names for IO pads should be self.

-########################################################

-#def make_port_sig($port)

- #if $port.get_parent().get_key() in ('pad_source', 'pad_sink')

- #set block = 'self'

- #set key = $flow_graph.get_pad_port_global_key($port)

- #else

- #set block = 'self.' + $port.get_parent().get_id()

- #set key = $port.get_key()

- #end if

- #if not $key.isdigit()

- #set key = repr($key)

- #end if

-($block, $key)#slurp

-#end def

-#if $connections

-

- $DIVIDER

- # Connections

- $DIVIDER

-#end if

-#for $con in $connections

- #set global $source = $con.get_source()

- #set global $sink = $con.get_sink()

- #include source=$connection_templates[($source.get_domain(), $sink.get_domain())]

-

-#end for

-########################################################

-## QT sink close method reimplementation

-########################################################

-#if $generate_options == 'qt_gui'

-

- def closeEvent(self, event):

- self.settings = Qt.QSettings("GNU Radio", "$class_name")

- self.settings.setValue("geometry", self.saveGeometry())

- event.accept()

- #if $flow_graph.get_option('qt_qss_theme')

-

- def setStyleSheetFromFile(self, filename):

- try:

- if not os.path.exists(filename):

- filename = os.path.join(

- gr.prefix(), "share", "gnuradio", "themes", filename)

- with open(filename) as ss:

- self.setStyleSheet(ss.read())

- except Exception as e:

- print >> sys.stderr, e

- #end if

-#end if

-########################################################

-##Create Callbacks

-## Write a set method for this variable that calls the callbacks

-########################################################

-#for $var in $parameters + $variables

-

- #set $id = $var.get_id()

- def get_$(id)(self):

- return self.$id

-

- def set_$(id)(self, $id):

- #if $flow_graph.get_option('thread_safe_setters')

- with self._lock:

- self.$id = $id

- #for $callback in $callbacks[$id]

- $indent($callback)

- #end for

- #else

- self.$id = $id

- #for $callback in $callbacks[$id]

- $indent($callback)

- #end for

- #end if

-#end for

-########################################################

-##Create Main

-## For top block code, generate a main routine.

-## Instantiate the top block and run as gui or cli.

-########################################################

-#def make_default($type, $param)

- #if $type == 'eng_float'

-eng_notation.num_to_str($param.get_make())#slurp

- #else

-$param.get_make()#slurp

- #end if

-#end def

-#def make_short_id($param)

- #set $short_id = $param.get_param('short_id').get_evaluated()

- #if $short_id

- #set $short_id = '-' + $short_id

- #end if

-$short_id#slurp

-#end def

-#if not $generate_options.startswith('hb')

-#set $params_eq_list = list()

-#if $parameters

-

-

-def argument_parser():

- #set $arg_parser_args = ''

- #if $flow_graph.get_option('description')

- #set $arg_parser_args = 'description=description'

- description = $repr($flow_graph.get_option('description'))

- #end if

- parser = ArgumentParser($arg_parser_args)

- #for $param in $parameters

- #set $type = $param.get_param('type').get_value()

- #if $type

- #silent $params_eq_list.append('%s=options.%s'%($param.get_id(), $param.get_id()))

- parser.add_argument(

- #if $make_short_id($param)

- "$make_short_id($param)", #slurp

- #end if

- "--$param.get_id().replace('_', '-')", dest="$param.get_id()", type=$type, default=$make_default($type, $param),

- help="Set $($param.get_param('label').get_evaluated() or $param.get_id()) [default=%(default)r]")

- #end if

- #end for

- return parser

-#end if

-

-

-def main(top_block_cls=$(class_name), options=None):

- #if $parameters

- if options is None:

- options = argument_parser().parse_args()

- #end if

- #if $flow_graph.get_option('realtime_scheduling')

- if gr.enable_realtime_scheduling() != gr.RT_OK:

- print "Error: failed to enable real-time scheduling."

- #end if

-

- #if $generate_options == 'qt_gui'

- if StrictVersion("4.5.0") <= StrictVersion(Qt.qVersion()) < StrictVersion("5.0.0"):

- style = gr.prefs().get_string('qtgui', 'style', 'raster')

- Qt.QApplication.setGraphicsSystem(style)

- qapp = Qt.QApplication(sys.argv)

-

- tb = top_block_cls($(', '.join($params_eq_list)))

- #if $flow_graph.get_option('run')

- #if $flow_graph.get_option('max_nouts')

- tb.start($flow_graph.get_option('max_nouts'))

- #else

- tb.start()

- #end if

- #end if

- #if $flow_graph.get_option('qt_qss_theme')

- tb.setStyleSheetFromFile($repr($flow_graph.get_option('qt_qss_theme')))

- #end if

- tb.show()

-

- def quitting():

- tb.stop()

- tb.wait()

- qapp.aboutToQuit.connect(quitting)

- #for $m in $monitors

- if $m.has_param('en'):

- if $m.get_param('en').get_value():

- (tb.$m.get_id()).start()

- else:

- sys.stderr.write("Monitor '{0}' does not have an enable ('en') parameter.".format("tb.$m.get_id()"))

- #end for

- qapp.exec_()

- #elif $generate_options == 'bokeh_gui'

- serverProc, port = bokehgui.utils.create_server()

- def killProc(signum, frame, tb):

- tb.stop()

- tb.wait()

- serverProc.terminate()

- serverProc.kill()

- time.sleep(1)

- try:

- \# Define the document instance

- doc = curdoc()

- #if $flow_graph.get_option('author')

- doc.title = "$title - $flow_graph.get_option('author')"

- #else

- doc.title = "$title"

- #end if

- session = push_session(doc, session_id="$flow_graph.get_option('id')",

- url = "http://localhost:" + port + "/bokehgui")

- \# Create Top Block instance

- tb = top_block_cls(doc)

- try:

- tb.start()

- signal.signal(signal.SIGTERM, functools.partial(killProc, tb=tb))

- session.loop_until_closed()

- finally:

- print "Exiting the simulation. Stopping Bokeh Server"

- tb.stop()

- tb.wait()

- finally:

- serverProc.terminate()

- serverProc.kill()

- #elif $generate_options == 'no_gui'

- tb = top_block_cls($(', '.join($params_eq_list)))

- #set $run_options = $flow_graph.get_option('run_options')

- #if $run_options == 'prompt'

- #if $flow_graph.get_option('max_nouts')

- tb.start($flow_graph.get_option('max_nouts'))

- #else

- tb.start()

- #end if

- #for $m in $monitors

- (tb.$m.get_id()).start()

- #end for

- try:

- raw_input('Press Enter to quit: ')

- except EOFError:

- pass

- tb.stop()

- #elif $run_options == 'run'

- #if $flow_graph.get_option('max_nouts')

- tb.start($flow_graph.get_option('max_nouts'))

- #else

- tb.start()

- #end if

- #end if

- #for $m in $monitors

- (tb.$m.get_id()).start()

- #end for

- tb.wait()

- #end if

-

-

-if __name__ == '__main__':

- main()

-#end if