summaryrefslogtreecommitdiff
path: root/grc/src/platforms/python/FlowGraph.py
diff options
context:
space:
mode:
Diffstat (limited to 'grc/src/platforms/python/FlowGraph.py')
-rw-r--r--grc/src/platforms/python/FlowGraph.py162
1 files changed, 0 insertions, 162 deletions
diff --git a/grc/src/platforms/python/FlowGraph.py b/grc/src/platforms/python/FlowGraph.py
deleted file mode 100644
index d0b997a58..000000000
--- a/grc/src/platforms/python/FlowGraph.py
+++ /dev/null
@@ -1,162 +0,0 @@
-"""
-Copyright 2008, 2009 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 utils import expr_utils
-from .. base.FlowGraph import FlowGraph as _FlowGraph
-from Block import Block
-from Connection import Connection
-
-def _get_value_expr(variable_block):
- """
- Get the expression to evaluate from the value param.
- Parameter blocks need to be evaluated so the stringify flag can be determined.
- @param variable_block the variable or parameter block
- @return the expression string
- """
- value_param = variable_block.get_param('value')
- if variable_block.get_key() == 'parameter': value_param.evaluate()
- return value_param.to_code()
-
-class FlowGraph(_FlowGraph):
-
- _eval_cache = dict()
- def _eval(self, code, namespace):
- """
- Evaluate the code with the given namespace.
- @param code a string with python code
- @param namespace a dict representing the namespace
- @return the resultant object
- """
- my_hash = hash(code + str(namespace))
- #cache if does not exist
- if not self._eval_cache.has_key(my_hash):
- self._eval_cache[my_hash] = eval(code, namespace, namespace)
- #return from cache
- return self._eval_cache[my_hash]
-
- def _get_io_signature(self, pad_key):
- """
- Get an io signature for this flow graph.
- The pad key determines the directionality of the io signature.
- @param pad_key a string of pad_source or pad_sink
- @return a dict with: type, nports, vlen, size
- """
- pads = filter(lambda b: b.get_key() == pad_key, self.get_enabled_blocks())
- if not pads: return {
- 'nports': '0',
- 'type': '',
- 'vlen': '0',
- 'size': '0',
- }
- pad = pads[0] #take only the first, user should not have more than 1
- #load io signature
- return {
- 'nports': str(pad.get_param('nports').evaluate()),
- 'type': str(pad.get_param('type').evaluate()),
- 'vlen': str(pad.get_param('vlen').evaluate()),
- 'size': pad.get_param('type').get_opt('size'),
- }
-
- def get_input_signature(self):
- """
- Get the io signature for the input side of this flow graph.
- The io signature with be "0", "0" if no pad source is present.
- @return a string tuple of type, num_ports, port_size
- """
- return self._get_io_signature('pad_source')
-
- def get_output_signature(self):
- """
- Get the io signature for the output side of this flow graph.
- The io signature with be "0", "0" if no pad sink is present.
- @return a string tuple of type, num_ports, port_size
- """
- return self._get_io_signature('pad_sink')
-
- def get_imports(self):
- """
- Get a set of all import statments in this flow graph namespace.
- @return a set of import statements
- """
- imports = sum([block.get_imports() for block in self.get_enabled_blocks()], [])
- imports = sorted(set(imports))
- return imports
-
- def get_variables(self):
- """
- Get a list of all variables in this flow graph namespace.
- Exclude paramterized variables.
- @return a sorted list of variable blocks in order of dependency (indep -> dep)
- """
- variables = filter(lambda b: b.get_key() in (
- 'variable', 'variable_slider', 'variable_chooser', 'variable_text_box'
- ), self.get_enabled_blocks())
- #map var id to variable block
- id2var = dict([(var.get_id(), var) for var in variables])
- #map var id to variable code
- #variable code is a concatenation of all param code (without the id param)
- id2expr = dict([(var.get_id(), var.get_param('value').get_value()) for var in variables])
- #sort according to dependency
- sorted_ids = expr_utils.sort_variables(id2expr)
- #create list of sorted variable blocks
- variables = [id2var[id] for id in sorted_ids]
- return variables
-
- def get_parameters(self):
- """
- Get a list of all paramterized variables in this flow graph namespace.
- @return a list of paramterized variables
- """
- parameters = filter(lambda b: b.get_key() == 'parameter', self.get_enabled_blocks())
- return parameters
-
- def evaluate(self, expr):
- """
- Evaluate the expression.
- @param expr the string expression
- @throw Exception bad expression
- @return the evaluated data
- """
- if self.is_flagged():
- self.deflag()
- #reload namespace
- n = dict()
- #load imports
- for imp in self.get_imports():
- try: exec imp in n
- except: pass
- #load parameters
- np = dict()
- for parameter in self.get_parameters():
- try:
- e = self._eval(_get_value_expr(parameter), n)
- np[parameter.get_id()] = e
- except: pass
- n.update(np) #merge param namespace
- #load variables
- for variable in self.get_variables():
- try:
- e = self._eval(_get_value_expr(variable), n)
- n[variable.get_id()] = e
- except: pass
- #make namespace public
- self.n = n
- #evaluate
- e = self._eval(expr, self.n)
- return e