Source code for openmdao.main.vartree

""" VariableTree class definition
import copy

# pylint: disable-msg=E0611,F0401
from enthought.traits.has_traits import FunctionType

from openmdao.main.variable import Variable
from openmdao.main.container import Container
from openmdao.main.datatypes.api import Slot, Str
from openmdao.main.rbac import rbac
from openmdao.main.mp_support import is_instance

[docs]class VariableTree(Container): """A container of variables with an input or output sense.""" _iotype = Str('') def __init__(self, iotype='', doc=None): super(VariableTree, self).__init__(doc=doc) self._iotype = iotype self.on_trait_change(self._iotype_modified, '_iotype') # register callbacks for our class traits for name, trait in self.class_traits().items(): if not name.startswith('_'): self.on_trait_change(self._trait_modified, name) @property
[docs] def iotype(self): return self._iotype
@rbac(('owner', 'user'))
[docs] def cpath_updated(self): if self.parent: if isinstance(self.parent, VariableTree): self._iotype = self.parent._iotype else: t = self.parent.trait( if t and t.iotype: self._iotype = t.iotype super(VariableTree, self).cpath_updated()
@rbac(('owner', 'user'))
[docs] def get_metadata(self, traitpath, metaname=None): if metaname == 'iotype': return self._iotype elif metaname is None: meta = super(VariableTree, self).get_metadata(traitpath, metaname) meta['iotype'] = self._iotype return meta else: return super(VariableTree, self).get_metadata(traitpath, metaname)
[docs] def copy(self): """Returns a deep copy of this VariableTree.""" return copy.deepcopy(self)
[docs] def add(self, name, obj): if isinstance(obj, VariableTree): if self.trait(name) is None: self.add_trait(name, Slot(VariableTree, iotype=obj._iotype)) self.on_trait_change(self._trait_modified, name) elif not isinstance(obj, Variable): msg = "a VariableTree may only contain Variables or other " + \ "VariableTrees" self.raise_exception(msg, TypeError) return super(VariableTree, self).add(name, obj)
[docs] def add_trait(self, name, trait): super(VariableTree, self).add_trait(name, trait) if not name.startswith('_'): self.on_trait_change(self._trait_modified, name)
[docs] def remove_trait(self, name): trait = self.get_trait(name) # remove the callback if trait: self.on_trait_change(self._trait_modified, name, remove=True) super(VariableTree, self).remove_trait(name)
[docs] def list_vars(self): """Return a list of Variables in this VariableTree.""" return [k for k in self.__dict__.keys() if not k.startswith('_')]
@rbac(('owner', 'user'))
[docs] def invalidate_deps(self, varnames=None, force=False): return None
def _iotype_modified(self, obj, name, old, new): for k, v in self.__dict__.items(): if isinstance(v, VariableTree) and v is not self.parent: v._iotype = new def _trait_modified(self, obj, name, old, new): # handle weird traits side-effect from hasattr call if name == 'trait_added': return if isinstance(new, VariableTree): obj = getattr(self, name) obj.parent = self obj._iotype = self._iotype if self._iotype == 'in': p = self while isinstance(p, VariableTree): vt = p p = p.parent # notify parent Component that this VariableTree has been modified if p is not None: t = p.trait( if t and t.iotype == 'in': p._input_trait_modified(p,, vt, vt)
[docs] def get_iotype(self, name): """Return the iotype of the Variable with the given name""" if self.get_trait(name) is None: self.raise_exception("'%s' not found" % name) return self._iotype
def _items(self, visited, recurse=False, **metadata): """Return an iterator that returns a list of tuples of the form (rel_pathname, obj) for each trait of this VariableTree that matches the given metadata. If recurse is True, also iterate through all child Containers of each Container found. """ if id(self) not in visited: visited.add(id(self)) if 'iotype' in metadata: meta_io = metadata['iotype'] matches_io = False if type( meta_io ) is FunctionType: if meta_io(self._iotype): matches_io = True elif meta_io == self._iotype: matches_io = True if matches_io: newdict = metadata.copy() del newdict['iotype'] else: matches_io = True newdict = metadata if matches_io: for name, trait in self._alltraits(**newdict).items(): if name.startswith('_'): continue obj = getattr(self, name) yield (name, obj) if recurse and is_instance(obj, VariableTree) and \ id(obj) not in visited: for chname, child in obj._items(visited, recurse, **metadata): yield ('.'.join([name, chname]), child)
[docs] def get_attributes(self, io_only=True): """ get attributes for this variable tree. Variables may also include slots. Used by the GUI. io_only: Bool Set to true if we only want to populate the input and output fields of the attributes dictionary.""" attrs = {} attrs['type'] = type(self).__name__ # Connection information found in parent comp's parent assy if not self.parent or not self.parent._parent or \ isinstance(self.parent, VariableTree): connected = [] else: graph = self.parent._parent._depgraph if self._iotype == 'in': connected = graph.get_connected_inputs() else: connected = graph.get_connected_outputs() variables = [] slots = [] for name in self.list_vars(): trait = self.get_trait(name) meta = self.get_metadata(name) value = getattr(self, name) ttype = trait.trait_type # Each variable type provides its own basic attributes attr, slot_attr = ttype.get_attribute(name, value, trait, meta) attr['connected'] = '' if name in connected: connections = self.parent._depgraph.connections_to(name) if self._iotype == 'in': # there can be only one connection to an input attr['connected'] = str([src for src, dst in \ connections]).replace('@xin.', '') else: attr['connected'] = str([dst for src, dst in \ connections]).replace('@xout.', '') variables.append(attr) # Process singleton and contained slots. if not io_only and slot_attr is not None: # We can hide slots (e.g., the Workflow slot in drivers) if 'hidden' not in meta or meta['hidden'] == False: slots.append(slot_attr) if self._iotype == 'in': panel = 'Inputs' else: panel = 'Outputs' attrs[panel] = variables attrs['Slots'] = slots return attrs # register a flattener for Cases
from import flatteners, flatten_obj def _flatten_vartree(name, vt): ret = [] for n, v in vt._items(set()): ret.extend([('.'.join([name, k]), v) for k, v in flatten_obj(n, v)]) return ret flatteners[VariableTree] = _flatten_vartree
OpenMDAO Home