Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/pydot.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
| author | shellac |
|---|---|
| date | Mon, 22 Mar 2021 18:12:50 +0000 |
| parents | |
| children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/pydot.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,1964 @@ +"""An interface to GraphViz.""" +from __future__ import division +from __future__ import print_function +import copy +import io +import errno +import os +import re +import subprocess +import sys +import tempfile +import warnings + +try: + import dot_parser +except Exception as e: + warnings.warn( + "`pydot` could not import `dot_parser`, " + "so `pydot` will be unable to parse DOT files. " + "The error was: {e}".format(e=e)) + + +__author__ = 'Ero Carrera' +__version__ = '1.4.2' +__license__ = 'MIT' + + +PY3 = sys.version_info >= (3, 0, 0) +if PY3: + str_type = str +else: + str_type = basestring + + +GRAPH_ATTRIBUTES = { 'Damping', 'K', 'URL', 'aspect', 'bb', 'bgcolor', + 'center', 'charset', 'clusterrank', 'colorscheme', 'comment', 'compound', + 'concentrate', 'defaultdist', 'dim', 'dimen', 'diredgeconstraints', + 'dpi', 'epsilon', 'esep', 'fontcolor', 'fontname', 'fontnames', + 'fontpath', 'fontsize', 'id', 'label', 'labeljust', 'labelloc', + 'landscape', 'layers', 'layersep', 'layout', 'levels', 'levelsgap', + 'lheight', 'lp', 'lwidth', 'margin', 'maxiter', 'mclimit', 'mindist', + 'mode', 'model', 'mosek', 'nodesep', 'nojustify', 'normalize', 'nslimit', + 'nslimit1', 'ordering', 'orientation', 'outputorder', 'overlap', + 'overlap_scaling', 'pack', 'packmode', 'pad', 'page', 'pagedir', + 'quadtree', 'quantum', 'rankdir', 'ranksep', 'ratio', 'remincross', + 'repulsiveforce', 'resolution', 'root', 'rotate', 'searchsize', 'sep', + 'showboxes', 'size', 'smoothing', 'sortv', 'splines', 'start', + 'stylesheet', 'target', 'truecolor', 'viewport', 'voro_margin', + # for subgraphs + 'rank' } + + +EDGE_ATTRIBUTES = { 'URL', 'arrowhead', 'arrowsize', 'arrowtail', + 'color', 'colorscheme', 'comment', 'constraint', 'decorate', 'dir', + 'edgeURL', 'edgehref', 'edgetarget', 'edgetooltip', 'fontcolor', + 'fontname', 'fontsize', 'headURL', 'headclip', 'headhref', 'headlabel', + 'headport', 'headtarget', 'headtooltip', 'href', 'id', 'label', + 'labelURL', 'labelangle', 'labeldistance', 'labelfloat', 'labelfontcolor', + 'labelfontname', 'labelfontsize', 'labelhref', 'labeltarget', + 'labeltooltip', 'layer', 'len', 'lhead', 'lp', 'ltail', 'minlen', + 'nojustify', 'penwidth', 'pos', 'samehead', 'sametail', 'showboxes', + 'style', 'tailURL', 'tailclip', 'tailhref', 'taillabel', 'tailport', + 'tailtarget', 'tailtooltip', 'target', 'tooltip', 'weight', + 'rank' } + + +NODE_ATTRIBUTES = { 'URL', 'color', 'colorscheme', 'comment', + 'distortion', 'fillcolor', 'fixedsize', 'fontcolor', 'fontname', + 'fontsize', 'group', 'height', 'id', 'image', 'imagescale', 'label', + 'labelloc', 'layer', 'margin', 'nojustify', 'orientation', 'penwidth', + 'peripheries', 'pin', 'pos', 'rects', 'regular', 'root', 'samplepoints', + 'shape', 'shapefile', 'showboxes', 'sides', 'skew', 'sortv', 'style', + 'target', 'tooltip', 'vertices', 'width', 'z', + # The following are attributes dot2tex + 'texlbl', 'texmode' } + + +CLUSTER_ATTRIBUTES = { 'K', 'URL', 'bgcolor', 'color', 'colorscheme', + 'fillcolor', 'fontcolor', 'fontname', 'fontsize', 'label', 'labeljust', + 'labelloc', 'lheight', 'lp', 'lwidth', 'nojustify', 'pencolor', + 'penwidth', 'peripheries', 'sortv', 'style', 'target', 'tooltip' } + + +DEFAULT_PROGRAMS = { + 'dot', + 'twopi', + 'neato', + 'circo', + 'fdp', + 'sfdp', +} + + +def is_windows(): + # type: () -> bool + return os.name == 'nt' + + +def is_anaconda(): + # type: () -> bool + import glob + return glob.glob(os.path.join(sys.prefix, 'conda-meta\\graphviz*.json')) != [] + + +def get_executable_extension(): + # type: () -> str + if is_windows(): + return '.bat' if is_anaconda() else '.exe' + else: + return '' + + +def call_graphviz(program, arguments, working_dir, **kwargs): + # explicitly inherit `$PATH`, on Windows too, + # with `shell=False` + + if program in DEFAULT_PROGRAMS: + extension = get_executable_extension() + program += extension + + if arguments is None: + arguments = [] + + env = { + 'PATH': os.environ.get('PATH', ''), + 'LD_LIBRARY_PATH': os.environ.get('LD_LIBRARY_PATH', ''), + 'SYSTEMROOT': os.environ.get('SYSTEMROOT', ''), + } + + program_with_args = [program, ] + arguments + + process = subprocess.Popen( + program_with_args, + env=env, + cwd=working_dir, + shell=False, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + **kwargs + ) + stdout_data, stderr_data = process.communicate() + + return stdout_data, stderr_data, process + + +# +# Extended version of ASPN's Python Cookbook Recipe: +# Frozen dictionaries. +# https://code.activestate.com/recipes/414283/ +# +# This version freezes dictionaries used as values within dictionaries. +# +class frozendict(dict): + def _blocked_attribute(obj): + raise AttributeError('A frozendict cannot be modified.') + _blocked_attribute = property(_blocked_attribute) + + __delitem__ = __setitem__ = clear = _blocked_attribute + pop = popitem = setdefault = update = _blocked_attribute + + def __new__(cls, *args, **kw): + new = dict.__new__(cls) + + args_ = [] + for arg in args: + if isinstance(arg, dict): + arg = copy.copy(arg) + for k in arg: + v = arg[k] + if isinstance(v, frozendict): + arg[k] = v + elif isinstance(v, dict): + arg[k] = frozendict(v) + elif isinstance(v, list): + v_ = list() + for elm in v: + if isinstance(elm, dict): + v_.append( frozendict(elm) ) + else: + v_.append( elm ) + arg[k] = tuple(v_) + args_.append( arg ) + else: + args_.append( arg ) + + dict.__init__(new, *args_, **kw) + return new + + def __init__(self, *args, **kw): + pass + + def __hash__(self): + try: + return self._cached_hash + except AttributeError: + h = self._cached_hash = hash(tuple(sorted(self.items()))) + return h + + def __repr__(self): + return "frozendict(%s)" % dict.__repr__(self) + + +dot_keywords = ['graph', 'subgraph', 'digraph', 'node', 'edge', 'strict'] + +id_re_alpha_nums = re.compile('^[_a-zA-Z][a-zA-Z0-9_,]*$', re.UNICODE) +id_re_alpha_nums_with_ports = re.compile( + '^[_a-zA-Z][a-zA-Z0-9_,:\"]*[a-zA-Z0-9_,\"]+$', re.UNICODE) +id_re_num = re.compile('^[0-9,]+$', re.UNICODE) +id_re_with_port = re.compile('^([^:]*):([^:]*)$', re.UNICODE) +id_re_dbl_quoted = re.compile('^\".*\"$', re.S|re.UNICODE) +id_re_html = re.compile('^<.*>$', re.S|re.UNICODE) + + +def needs_quotes( s ): + """Checks whether a string is a dot language ID. + + It will check whether the string is solely composed + by the characters allowed in an ID or not. + If the string is one of the reserved keywords it will + need quotes too but the user will need to add them + manually. + """ + + # If the name is a reserved keyword it will need quotes but pydot + # can't tell when it's being used as a keyword or when it's simply + # a name. Hence the user needs to supply the quotes when an element + # would use a reserved keyword as name. This function will return + # false indicating that a keyword string, if provided as-is, won't + # need quotes. + if s in dot_keywords: + return False + + chars = [ord(c) for c in s if ord(c)>0x7f or ord(c)==0] + if chars and not id_re_dbl_quoted.match(s) and not id_re_html.match(s): + return True + + for test_re in [id_re_alpha_nums, id_re_num, + id_re_dbl_quoted, id_re_html, + id_re_alpha_nums_with_ports]: + if test_re.match(s): + return False + + m = id_re_with_port.match(s) + if m: + return needs_quotes(m.group(1)) or needs_quotes(m.group(2)) + + return True + + +def quote_if_necessary(s): + """Enclose attribute value in quotes, if needed.""" + if isinstance(s, bool): + if s is True: + return 'True' + return 'False' + + if not isinstance( s, str_type): + return s + + if not s: + return s + + if needs_quotes(s): + replace = {'"' : r'\"', + "\n" : r'\n', + "\r" : r'\r'} + for (a,b) in replace.items(): + s = s.replace(a, b) + + return '"' + s + '"' + + return s + + + +def graph_from_dot_data(s): + """Load graphs from DOT description in string `s`. + + @param s: string in [DOT language]( + https://en.wikipedia.org/wiki/DOT_(graph_description_language)) + + @return: Graphs that result from parsing. + @rtype: `list` of `pydot.Dot` + """ + return dot_parser.parse_dot_data(s) + + +def graph_from_dot_file(path, encoding=None): + """Load graphs from DOT file at `path`. + + @param path: to DOT file + @param encoding: as passed to `io.open`. + For example, `'utf-8'`. + + @return: Graphs that result from parsing. + @rtype: `list` of `pydot.Dot` + """ + with io.open(path, 'rt', encoding=encoding) as f: + s = f.read() + if not PY3: + s = unicode(s) + graphs = graph_from_dot_data(s) + return graphs + + + +def graph_from_edges(edge_list, node_prefix='', directed=False): + """Creates a basic graph out of an edge list. + + The edge list has to be a list of tuples representing + the nodes connected by the edge. + The values can be anything: bool, int, float, str. + + If the graph is undirected by default, it is only + calculated from one of the symmetric halves of the matrix. + """ + + if directed: + graph = Dot(graph_type='digraph') + + else: + graph = Dot(graph_type='graph') + + for edge in edge_list: + + if isinstance(edge[0], str): + src = node_prefix + edge[0] + else: + src = node_prefix + str(edge[0]) + + if isinstance(edge[1], str): + dst = node_prefix + edge[1] + else: + dst = node_prefix + str(edge[1]) + + e = Edge( src, dst ) + graph.add_edge(e) + + return graph + + +def graph_from_adjacency_matrix(matrix, node_prefix= u'', directed=False): + """Creates a basic graph out of an adjacency matrix. + + The matrix has to be a list of rows of values + representing an adjacency matrix. + The values can be anything: bool, int, float, as long + as they can evaluate to True or False. + """ + + node_orig = 1 + + if directed: + graph = Dot(graph_type='digraph') + else: + graph = Dot(graph_type='graph') + + for row in matrix: + if not directed: + skip = matrix.index(row) + r = row[skip:] + else: + skip = 0 + r = row + node_dest = skip+1 + + for e in r: + if e: + graph.add_edge( + Edge('%s%s' % (node_prefix, node_orig), + '%s%s' % (node_prefix, node_dest))) + node_dest += 1 + node_orig += 1 + + return graph + + +def graph_from_incidence_matrix(matrix, node_prefix='', directed=False): + """Creates a basic graph out of an incidence matrix. + + The matrix has to be a list of rows of values + representing an incidence matrix. + The values can be anything: bool, int, float, as long + as they can evaluate to True or False. + """ + + node_orig = 1 + + if directed: + graph = Dot(graph_type='digraph') + else: + graph = Dot(graph_type='graph') + + for row in matrix: + nodes = [] + c = 1 + + for node in row: + if node: + nodes.append(c*node) + c += 1 + nodes.sort() + + if len(nodes) == 2: + graph.add_edge( + Edge('%s%s' % (node_prefix, abs(nodes[0])), + '%s%s' % (node_prefix, nodes[1]))) + + if not directed: + graph.set_simplify(True) + + return graph + + +class Common(object): + """Common information to several classes. + + Should not be directly used, several classes are derived from + this one. + """ + + + def __getstate__(self): + + dict = copy.copy(self.obj_dict) + + return dict + + + def __setstate__(self, state): + + self.obj_dict = state + + + def __get_attribute__(self, attr): + """Look for default attributes for this node""" + + attr_val = self.obj_dict['attributes'].get(attr, None) + + if attr_val is None: + # get the defaults for nodes/edges + + default_node_name = self.obj_dict['type'] + + # The defaults for graphs are set on a node named 'graph' + if default_node_name in ('subgraph', 'digraph', 'cluster'): + default_node_name = 'graph' + + g = self.get_parent_graph() + if g is not None: + defaults = g.get_node( default_node_name ) + else: + return None + + # Multiple defaults could be set by having repeated 'graph [...]' + # 'node [...]', 'edge [...]' statements. In such case, if the + # same attribute is set in different statements, only the first + # will be returned. In order to get all, one would call the + # get_*_defaults() methods and handle those. Or go node by node + # (of the ones specifying defaults) and modify the attributes + # individually. + # + if not isinstance(defaults, (list, tuple)): + defaults = [defaults] + + for default in defaults: + attr_val = default.obj_dict['attributes'].get(attr, None) + if attr_val: + return attr_val + else: + return attr_val + + return None + + + def set_parent_graph(self, parent_graph): + + self.obj_dict['parent_graph'] = parent_graph + + + def get_parent_graph(self): + + return self.obj_dict.get('parent_graph', None) + + + def set(self, name, value): + """Set an attribute value by name. + + Given an attribute 'name' it will set its value to 'value'. + There's always the possibility of using the methods: + + set_'name'(value) + + which are defined for all the existing attributes. + """ + + self.obj_dict['attributes'][name] = value + + + def get(self, name): + """Get an attribute value by name. + + Given an attribute 'name' it will get its value. + There's always the possibility of using the methods: + + get_'name'() + + which are defined for all the existing attributes. + """ + + return self.obj_dict['attributes'].get(name, None) + + + def get_attributes(self): + """""" + + return self.obj_dict['attributes'] + + + def set_sequence(self, seq): + + self.obj_dict['sequence'] = seq + + + def get_sequence(self): + + return self.obj_dict['sequence'] + + + def create_attribute_methods(self, obj_attributes): + + #for attr in self.obj_dict['attributes']: + for attr in obj_attributes: + + # Generate all the Setter methods. + # + self.__setattr__( + 'set_'+attr, + lambda x, a=attr : + self.obj_dict['attributes'].__setitem__(a, x) ) + + # Generate all the Getter methods. + # + self.__setattr__( + 'get_'+attr, lambda a=attr : self.__get_attribute__(a)) + + + +class Error(Exception): + """General error handling class. + """ + def __init__(self, value): + self.value = value + def __str__(self): + return self.value + + +class InvocationException(Exception): + """Indicate problem while running any GraphViz executable. + """ + def __init__(self, value): + self.value = value + def __str__(self): + return self.value + + + +class Node(Common): + """A graph node. + + This class represents a graph's node with all its attributes. + + node(name, attribute=value, ...) + + name: node's name + + All the attributes defined in the Graphviz dot language should + be supported. + """ + + def __init__(self, name = '', obj_dict = None, **attrs): + + # + # Nodes will take attributes of + # all other types because the defaults + # for any GraphViz object are dealt with + # as if they were Node definitions + # + + if obj_dict is not None: + + self.obj_dict = obj_dict + + else: + + self.obj_dict = dict() + + # Copy the attributes + # + self.obj_dict[ 'attributes' ] = dict( attrs ) + self.obj_dict[ 'type' ] = 'node' + self.obj_dict[ 'parent_graph' ] = None + self.obj_dict[ 'parent_node_list' ] = None + self.obj_dict[ 'sequence' ] = None + + # Remove the compass point + # + port = None + if isinstance(name, str_type) and not name.startswith('"'): + idx = name.find(':') + if idx > 0 and idx+1 < len(name): + name, port = name[:idx], name[idx:] + + if isinstance(name, int): + name = str(name) + + self.obj_dict['name'] = quote_if_necessary(name) + self.obj_dict['port'] = port + + self.create_attribute_methods(NODE_ATTRIBUTES) + + def __str__(self): + return self.to_string() + + + def set_name(self, node_name): + """Set the node's name.""" + + self.obj_dict['name'] = node_name + + + def get_name(self): + """Get the node's name.""" + + return self.obj_dict['name'] + + + def get_port(self): + """Get the node's port.""" + + return self.obj_dict['port'] + + + def add_style(self, style): + + styles = self.obj_dict['attributes'].get('style', None) + if not styles and style: + styles = [ style ] + else: + styles = styles.split(',') + styles.append( style ) + + self.obj_dict['attributes']['style'] = ','.join( styles ) + + + def to_string(self): + """Return string representation of node in DOT language.""" + + + # RMF: special case defaults for node, edge and graph properties. + # + node = quote_if_necessary(self.obj_dict['name']) + + node_attr = list() + + for attr in sorted(self.obj_dict['attributes']): + value = self.obj_dict['attributes'][attr] + if value == '': + value = '""' + if value is not None: + node_attr.append( + '%s=%s' % (attr, quote_if_necessary(value) ) ) + else: + node_attr.append( attr ) + + + # No point in having nodes setting any defaults if the don't set + # any attributes... + # + if node in ('graph', 'node', 'edge') and len(node_attr) == 0: + return '' + + node_attr = ', '.join(node_attr) + + if node_attr: + node += ' [' + node_attr + ']' + + return node + ';' + + + +class Edge(Common): + """A graph edge. + + This class represents a graph's edge with all its attributes. + + edge(src, dst, attribute=value, ...) + + src: source node, subgraph or cluster + dst: destination node, subgraph or cluster + + `src` and `dst` can be specified as a `Node`, `Subgraph` or + `Cluster` object, or as the name string of such a component. + + All the attributes defined in the Graphviz dot language should + be supported. + + Attributes can be set through the dynamically generated methods: + + set_[attribute name], i.e. set_label, set_fontname + + or directly by using the instance's special dictionary: + + Edge.obj_dict['attributes'][attribute name], i.e. + + edge_instance.obj_dict['attributes']['label'] + edge_instance.obj_dict['attributes']['fontname'] + + """ + + def __init__(self, src='', dst='', obj_dict=None, **attrs): + self.obj_dict = dict() + if isinstance(src, (Node, Subgraph, Cluster)): + src = src.get_name() + if isinstance(dst, (Node, Subgraph, Cluster)): + dst = dst.get_name() + points = (quote_if_necessary(src), + quote_if_necessary(dst)) + self.obj_dict['points'] = points + if obj_dict is None: + # Copy the attributes + self.obj_dict[ 'attributes' ] = dict( attrs ) + self.obj_dict[ 'type' ] = 'edge' + self.obj_dict[ 'parent_graph' ] = None + self.obj_dict[ 'parent_edge_list' ] = None + self.obj_dict[ 'sequence' ] = None + else: + self.obj_dict = obj_dict + self.create_attribute_methods(EDGE_ATTRIBUTES) + + def __str__(self): + return self.to_string() + + + def get_source(self): + """Get the edges source node name.""" + + return self.obj_dict['points'][0] + + + def get_destination(self): + """Get the edge's destination node name.""" + + return self.obj_dict['points'][1] + + + def __hash__(self): + + return hash( hash(self.get_source()) + + hash(self.get_destination()) ) + + + def __eq__(self, edge): + """Compare two edges. + + If the parent graph is directed, arcs linking + node A to B are considered equal and A->B != B->A + + If the parent graph is undirected, any edge + connecting two nodes is equal to any other + edge connecting the same nodes, A->B == B->A + """ + + if not isinstance(edge, Edge): + raise Error('Can not compare and ' + 'edge to a non-edge object.') + + if self.get_parent_graph().get_top_graph_type() == 'graph': + + # If the graph is undirected, the edge has neither + # source nor destination. + # + if ( ( self.get_source() == edge.get_source() and + self.get_destination() == edge.get_destination() ) or + ( edge.get_source() == self.get_destination() and + edge.get_destination() == self.get_source() ) ): + return True + + else: + + if (self.get_source()==edge.get_source() and + self.get_destination()==edge.get_destination()): + return True + + return False + + if not PY3: + def __ne__(self, other): + result = self.__eq__(other) + if result is NotImplemented: + return NotImplemented + return not result + + def parse_node_ref(self, node_str): + + if not isinstance(node_str, str): + return node_str + + if node_str.startswith('"') and node_str.endswith('"'): + + return node_str + + node_port_idx = node_str.rfind(':') + + if (node_port_idx>0 and node_str[0]=='"' and + node_str[node_port_idx-1]=='"'): + + return node_str + + if node_port_idx>0: + + a = node_str[:node_port_idx] + b = node_str[node_port_idx+1:] + + node = quote_if_necessary(a) + + node += ':'+quote_if_necessary(b) + + return node + + return node_str + + + def to_string(self): + """Return string representation of edge in DOT language.""" + + src = self.parse_node_ref( self.get_source() ) + dst = self.parse_node_ref( self.get_destination() ) + + if isinstance(src, frozendict): + edge = [ Subgraph(obj_dict=src).to_string() ] + elif isinstance(src, int): + edge = [ str(src) ] + else: + edge = [ src ] + + if (self.get_parent_graph() and + self.get_parent_graph().get_top_graph_type() and + self.get_parent_graph().get_top_graph_type() == 'digraph' ): + + edge.append( '->' ) + + else: + edge.append( '--' ) + + if isinstance(dst, frozendict): + edge.append( Subgraph(obj_dict=dst).to_string() ) + elif isinstance(dst, int): + edge.append( str(dst) ) + else: + edge.append( dst ) + + + edge_attr = list() + + for attr in sorted(self.obj_dict['attributes']): + value = self.obj_dict['attributes'][attr] + if value == '': + value = '""' + if value is not None: + edge_attr.append( + '%s=%s' % (attr, quote_if_necessary(value) ) ) + else: + edge_attr.append( attr ) + + edge_attr = ', '.join(edge_attr) + + if edge_attr: + edge.append( ' [' + edge_attr + ']' ) + + return ' '.join(edge) + ';' + + + + + +class Graph(Common): + """Class representing a graph in Graphviz's dot language. + + This class implements the methods to work on a representation + of a graph in Graphviz's dot language. + + graph( graph_name='G', graph_type='digraph', + strict=False, suppress_disconnected=False, attribute=value, ...) + + graph_name: + the graph's name + graph_type: + can be 'graph' or 'digraph' + suppress_disconnected: + defaults to False, which will remove from the + graph any disconnected nodes. + simplify: + if True it will avoid displaying equal edges, i.e. + only one edge between two nodes. removing the + duplicated ones. + + All the attributes defined in the Graphviz dot language should + be supported. + + Attributes can be set through the dynamically generated methods: + + set_[attribute name], i.e. set_size, set_fontname + + or using the instance's attributes: + + Graph.obj_dict['attributes'][attribute name], i.e. + + graph_instance.obj_dict['attributes']['label'] + graph_instance.obj_dict['attributes']['fontname'] + """ + + + def __init__(self, graph_name='G', obj_dict=None, + graph_type='digraph', strict=False, + suppress_disconnected=False, simplify=False, **attrs): + + if obj_dict is not None: + self.obj_dict = obj_dict + + else: + + self.obj_dict = dict() + + self.obj_dict['attributes'] = dict(attrs) + + if graph_type not in ['graph', 'digraph']: + raise Error(( + 'Invalid type "{t}". ' + 'Accepted graph types are: ' + 'graph, digraph').format(t=graph_type)) + + + self.obj_dict['name'] = quote_if_necessary(graph_name) + self.obj_dict['type'] = graph_type + + self.obj_dict['strict'] = strict + self.obj_dict['suppress_disconnected'] = suppress_disconnected + self.obj_dict['simplify'] = simplify + + self.obj_dict['current_child_sequence'] = 1 + self.obj_dict['nodes'] = dict() + self.obj_dict['edges'] = dict() + self.obj_dict['subgraphs'] = dict() + + self.set_parent_graph(self) + + + self.create_attribute_methods(GRAPH_ATTRIBUTES) + + def __str__(self): + return self.to_string() + + + def get_graph_type(self): + + return self.obj_dict['type'] + + + def get_top_graph_type(self): + + parent = self + while True: + parent_ = parent.get_parent_graph() + if parent_ == parent: + break + parent = parent_ + + return parent.obj_dict['type'] + + + def set_graph_defaults(self, **attrs): + + self.add_node( Node('graph', **attrs) ) + + + def get_graph_defaults(self, **attrs): + + graph_nodes = self.get_node('graph') + + if isinstance( graph_nodes, (list, tuple)): + return [ node.get_attributes() for node in graph_nodes ] + + return graph_nodes.get_attributes() + + + + def set_node_defaults(self, **attrs): + """Define default node attributes. + + These attributes only apply to nodes added to the graph after + calling this method. + """ + self.add_node( Node('node', **attrs) ) + + + def get_node_defaults(self, **attrs): + + + graph_nodes = self.get_node('node') + + if isinstance( graph_nodes, (list, tuple)): + return [ node.get_attributes() for node in graph_nodes ] + + return graph_nodes.get_attributes() + + + def set_edge_defaults(self, **attrs): + + self.add_node( Node('edge', **attrs) ) + + + + def get_edge_defaults(self, **attrs): + + graph_nodes = self.get_node('edge') + + if isinstance( graph_nodes, (list, tuple)): + return [ node.get_attributes() for node in graph_nodes ] + + return graph_nodes.get_attributes() + + + + def set_simplify(self, simplify): + """Set whether to simplify or not. + + If True it will avoid displaying equal edges, i.e. + only one edge between two nodes. removing the + duplicated ones. + """ + + self.obj_dict['simplify'] = simplify + + + + def get_simplify(self): + """Get whether to simplify or not. + + Refer to set_simplify for more information. + """ + + return self.obj_dict['simplify'] + + + def set_type(self, graph_type): + """Set the graph's type, 'graph' or 'digraph'.""" + + self.obj_dict['type'] = graph_type + + + + def get_type(self): + """Get the graph's type, 'graph' or 'digraph'.""" + + return self.obj_dict['type'] + + + + def set_name(self, graph_name): + """Set the graph's name.""" + + self.obj_dict['name'] = graph_name + + + + def get_name(self): + """Get the graph's name.""" + + return self.obj_dict['name'] + + + + def set_strict(self, val): + """Set graph to 'strict' mode. + + This option is only valid for top level graphs. + """ + + self.obj_dict['strict'] = val + + + + def get_strict(self, val): + """Get graph's 'strict' mode (True, False). + + This option is only valid for top level graphs. + """ + + return self.obj_dict['strict'] + + + + def set_suppress_disconnected(self, val): + """Suppress disconnected nodes in the output graph. + + This option will skip nodes in + the graph with no incoming or outgoing + edges. This option works also + for subgraphs and has effect only in the + current graph/subgraph. + """ + + self.obj_dict['suppress_disconnected'] = val + + + + def get_suppress_disconnected(self, val): + """Get if suppress disconnected is set. + + Refer to set_suppress_disconnected for more information. + """ + + return self.obj_dict['suppress_disconnected'] + + + def get_next_sequence_number(self): + + seq = self.obj_dict['current_child_sequence'] + + self.obj_dict['current_child_sequence'] += 1 + + return seq + + + + def add_node(self, graph_node): + """Adds a node object to the graph. + + It takes a node object as its only argument and returns + None. + """ + + if not isinstance(graph_node, Node): + raise TypeError( + 'add_node() received ' + + 'a non node class object: ' + str(graph_node)) + + + node = self.get_node(graph_node.get_name()) + + if not node: + + self.obj_dict['nodes'][graph_node.get_name()] = [ + graph_node.obj_dict ] + + #self.node_dict[graph_node.get_name()] = graph_node.attributes + graph_node.set_parent_graph(self.get_parent_graph()) + + else: + + self.obj_dict['nodes'][graph_node.get_name()].append( + graph_node.obj_dict ) + + graph_node.set_sequence(self.get_next_sequence_number()) + + + + def del_node(self, name, index=None): + """Delete a node from the graph. + + Given a node's name all node(s) with that same name + will be deleted if 'index' is not specified or set + to None. + If there are several nodes with that same name and + 'index' is given, only the node in that position + will be deleted. + + 'index' should be an integer specifying the position + of the node to delete. If index is larger than the + number of nodes with that name, no action is taken. + + If nodes are deleted it returns True. If no action + is taken it returns False. + """ + + if isinstance(name, Node): + name = name.get_name() + + if name in self.obj_dict['nodes']: + + if (index is not None and + index < len(self.obj_dict['nodes'][name])): + del self.obj_dict['nodes'][name][index] + return True + else: + del self.obj_dict['nodes'][name] + return True + + return False + + + def get_node(self, name): + """Retrieve a node from the graph. + + Given a node's name the corresponding Node + instance will be returned. + + If one or more nodes exist with that name a list of + Node instances is returned. + An empty list is returned otherwise. + """ + + match = list() + + if name in self.obj_dict['nodes']: + + match.extend( + [Node(obj_dict=obj_dict) + for obj_dict in self.obj_dict['nodes'][name]]) + + return match + + + def get_nodes(self): + """Get the list of Node instances.""" + + return self.get_node_list() + + + def get_node_list(self): + """Get the list of Node instances. + + This method returns the list of Node instances + composing the graph. + """ + + node_objs = list() + + for node in self.obj_dict['nodes']: + obj_dict_list = self.obj_dict['nodes'][node] + node_objs.extend( [ Node( obj_dict = obj_d ) + for obj_d in obj_dict_list ] ) + + return node_objs + + + + def add_edge(self, graph_edge): + """Adds an edge object to the graph. + + It takes a edge object as its only argument and returns + None. + """ + + if not isinstance(graph_edge, Edge): + raise TypeError( + 'add_edge() received a non edge class object: ' + + str(graph_edge)) + + edge_points = ( graph_edge.get_source(), + graph_edge.get_destination() ) + + if edge_points in self.obj_dict['edges']: + + edge_list = self.obj_dict['edges'][edge_points] + edge_list.append(graph_edge.obj_dict) + + else: + + self.obj_dict['edges'][edge_points] = [ graph_edge.obj_dict ] + + + graph_edge.set_sequence( self.get_next_sequence_number() ) + + graph_edge.set_parent_graph( self.get_parent_graph() ) + + + + def del_edge(self, src_or_list, dst=None, index=None): + """Delete an edge from the graph. + + Given an edge's (source, destination) node names all + matching edges(s) will be deleted if 'index' is not + specified or set to None. + If there are several matching edges and 'index' is + given, only the edge in that position will be deleted. + + 'index' should be an integer specifying the position + of the edge to delete. If index is larger than the + number of matching edges, no action is taken. + + If edges are deleted it returns True. If no action + is taken it returns False. + """ + + if isinstance( src_or_list, (list, tuple)): + if dst is not None and isinstance(dst, int): + index = dst + src, dst = src_or_list + else: + src, dst = src_or_list, dst + + if isinstance(src, Node): + src = src.get_name() + + if isinstance(dst, Node): + dst = dst.get_name() + + if (src, dst) in self.obj_dict['edges']: + + if (index is not None and + index < len(self.obj_dict['edges'][(src, dst)])): + del self.obj_dict['edges'][(src, dst)][index] + return True + else: + del self.obj_dict['edges'][(src, dst)] + return True + + return False + + + def get_edge(self, src_or_list, dst=None): + """Retrieved an edge from the graph. + + Given an edge's source and destination the corresponding + Edge instance(s) will be returned. + + If one or more edges exist with that source and destination + a list of Edge instances is returned. + An empty list is returned otherwise. + """ + + if isinstance( src_or_list, (list, tuple)) and dst is None: + edge_points = tuple(src_or_list) + edge_points_reverse = (edge_points[1], edge_points[0]) + else: + edge_points = (src_or_list, dst) + edge_points_reverse = (dst, src_or_list) + + match = list() + + if edge_points in self.obj_dict['edges'] or ( + self.get_top_graph_type() == 'graph' and + edge_points_reverse in self.obj_dict['edges']): + + edges_obj_dict = self.obj_dict['edges'].get( + edge_points, + self.obj_dict['edges'].get( edge_points_reverse, None )) + + for edge_obj_dict in edges_obj_dict: + match.append( + Edge(edge_points[0], + edge_points[1], + obj_dict=edge_obj_dict)) + + return match + + + def get_edges(self): + return self.get_edge_list() + + + def get_edge_list(self): + """Get the list of Edge instances. + + This method returns the list of Edge instances + composing the graph. + """ + + edge_objs = list() + + for edge in self.obj_dict['edges']: + obj_dict_list = self.obj_dict['edges'][edge] + edge_objs.extend( + [Edge(obj_dict=obj_d) + for obj_d in obj_dict_list]) + + return edge_objs + + + + def add_subgraph(self, sgraph): + """Adds an subgraph object to the graph. + + It takes a subgraph object as its only argument and returns + None. + """ + + if (not isinstance(sgraph, Subgraph) and + not isinstance(sgraph, Cluster)): + raise TypeError( + 'add_subgraph() received a non subgraph class object:' + + str(sgraph)) + + if sgraph.get_name() in self.obj_dict['subgraphs']: + + sgraph_list = self.obj_dict['subgraphs'][ sgraph.get_name() ] + sgraph_list.append( sgraph.obj_dict ) + + else: + self.obj_dict['subgraphs'][sgraph.get_name()] = [ + sgraph.obj_dict] + + sgraph.set_sequence( self.get_next_sequence_number() ) + + sgraph.set_parent_graph( self.get_parent_graph() ) + + + + + def get_subgraph(self, name): + """Retrieved a subgraph from the graph. + + Given a subgraph's name the corresponding + Subgraph instance will be returned. + + If one or more subgraphs exist with the same name, a list of + Subgraph instances is returned. + An empty list is returned otherwise. + """ + + match = list() + + if name in self.obj_dict['subgraphs']: + + sgraphs_obj_dict = self.obj_dict['subgraphs'].get( name ) + + for obj_dict_list in sgraphs_obj_dict: + #match.extend( Subgraph( obj_dict = obj_d ) + # for obj_d in obj_dict_list ) + match.append( Subgraph( obj_dict = obj_dict_list ) ) + + return match + + + def get_subgraphs(self): + + return self.get_subgraph_list() + + + def get_subgraph_list(self): + """Get the list of Subgraph instances. + + This method returns the list of Subgraph instances + in the graph. + """ + + sgraph_objs = list() + + for sgraph in self.obj_dict['subgraphs']: + obj_dict_list = self.obj_dict['subgraphs'][sgraph] + sgraph_objs.extend( + [Subgraph(obj_dict=obj_d) + for obj_d in obj_dict_list]) + + return sgraph_objs + + + + def set_parent_graph(self, parent_graph): + + self.obj_dict['parent_graph'] = parent_graph + + for k in self.obj_dict['nodes']: + obj_list = self.obj_dict['nodes'][k] + for obj in obj_list: + obj['parent_graph'] = parent_graph + + for k in self.obj_dict['edges']: + obj_list = self.obj_dict['edges'][k] + for obj in obj_list: + obj['parent_graph'] = parent_graph + + for k in self.obj_dict['subgraphs']: + obj_list = self.obj_dict['subgraphs'][k] + for obj in obj_list: + Graph(obj_dict=obj).set_parent_graph(parent_graph) + + + + def to_string(self): + """Return string representation of graph in DOT language. + + @return: graph and subelements + @rtype: `str` + """ + + + graph = list() + + if self.obj_dict.get('strict', None) is not None: + + if (self == self.get_parent_graph() and + self.obj_dict['strict']): + + graph.append('strict ') + + graph_type = self.obj_dict['type'] + if (graph_type == 'subgraph' and + not self.obj_dict.get('show_keyword', True)): + graph_type = '' + s = '{type} {name} {{\n'.format( + type=graph_type, + name=self.obj_dict['name']) + graph.append(s) + + for attr in sorted(self.obj_dict['attributes']): + + if self.obj_dict['attributes'].get(attr, None) is not None: + + val = self.obj_dict['attributes'].get(attr) + if val == '': + val = '""' + if val is not None: + graph.append('%s=%s' % + (attr, quote_if_necessary(val))) + else: + graph.append( attr ) + + graph.append( ';\n' ) + + + edges_done = set() + + edge_obj_dicts = list() + for k in self.obj_dict['edges']: + edge_obj_dicts.extend(self.obj_dict['edges'][k]) + + if edge_obj_dicts: + edge_src_set, edge_dst_set = list(zip( + *[obj['points'] for obj in edge_obj_dicts])) + edge_src_set, edge_dst_set = set(edge_src_set), set(edge_dst_set) + else: + edge_src_set, edge_dst_set = set(), set() + + node_obj_dicts = list() + for k in self.obj_dict['nodes']: + node_obj_dicts.extend(self.obj_dict['nodes'][k]) + + sgraph_obj_dicts = list() + for k in self.obj_dict['subgraphs']: + sgraph_obj_dicts.extend(self.obj_dict['subgraphs'][k]) + + + obj_list = [(obj['sequence'], obj) + for obj in (edge_obj_dicts + + node_obj_dicts + sgraph_obj_dicts) ] + obj_list.sort(key=lambda x: x[0]) + + for idx, obj in obj_list: + + if obj['type'] == 'node': + + node = Node(obj_dict=obj) + + if self.obj_dict.get('suppress_disconnected', False): + + if (node.get_name() not in edge_src_set and + node.get_name() not in edge_dst_set): + + continue + + graph.append( node.to_string()+'\n' ) + + elif obj['type'] == 'edge': + + edge = Edge(obj_dict=obj) + + if (self.obj_dict.get('simplify', False) and + edge in edges_done): + continue + + graph.append( edge.to_string() + '\n' ) + edges_done.add(edge) + + else: + + sgraph = Subgraph(obj_dict=obj) + + graph.append( sgraph.to_string()+'\n' ) + + graph.append( '}\n' ) + + return ''.join(graph) + + + +class Subgraph(Graph): + + """Class representing a subgraph in Graphviz's dot language. + + This class implements the methods to work on a representation + of a subgraph in Graphviz's dot language. + + subgraph(graph_name='subG', + suppress_disconnected=False, + attribute=value, + ...) + + graph_name: + the subgraph's name + suppress_disconnected: + defaults to false, which will remove from the + subgraph any disconnected nodes. + All the attributes defined in the Graphviz dot language should + be supported. + + Attributes can be set through the dynamically generated methods: + + set_[attribute name], i.e. set_size, set_fontname + + or using the instance's attributes: + + Subgraph.obj_dict['attributes'][attribute name], i.e. + + subgraph_instance.obj_dict['attributes']['label'] + subgraph_instance.obj_dict['attributes']['fontname'] + """ + + + # RMF: subgraph should have all the + # attributes of graph so it can be passed + # as a graph to all methods + # + def __init__(self, graph_name='', + obj_dict=None, suppress_disconnected=False, + simplify=False, **attrs): + + + Graph.__init__( + self, graph_name=graph_name, obj_dict=obj_dict, + suppress_disconnected=suppress_disconnected, + simplify=simplify, **attrs) + + if obj_dict is None: + + self.obj_dict['type'] = 'subgraph' + + + + +class Cluster(Graph): + + """Class representing a cluster in Graphviz's dot language. + + This class implements the methods to work on a representation + of a cluster in Graphviz's dot language. + + cluster(graph_name='subG', + suppress_disconnected=False, + attribute=value, + ...) + + graph_name: + the cluster's name + (the string 'cluster' will be always prepended) + suppress_disconnected: + defaults to false, which will remove from the + cluster any disconnected nodes. + All the attributes defined in the Graphviz dot language should + be supported. + + Attributes can be set through the dynamically generated methods: + + set_[attribute name], i.e. set_color, set_fontname + + or using the instance's attributes: + + Cluster.obj_dict['attributes'][attribute name], i.e. + + cluster_instance.obj_dict['attributes']['label'] + cluster_instance.obj_dict['attributes']['fontname'] + """ + + + def __init__(self, graph_name='subG', + obj_dict=None, suppress_disconnected=False, + simplify=False, **attrs): + + Graph.__init__( + self, graph_name=graph_name, obj_dict=obj_dict, + suppress_disconnected=suppress_disconnected, + simplify=simplify, **attrs) + + if obj_dict is None: + + self.obj_dict['type'] = 'subgraph' + self.obj_dict['name'] = quote_if_necessary('cluster_'+graph_name) + + self.create_attribute_methods(CLUSTER_ATTRIBUTES) + + + + + + +class Dot(Graph): + """A container for handling a dot language file. + + This class implements methods to write and process + a dot language file. It is a derived class of + the base class 'Graph'. + """ + + + + def __init__(self, *argsl, **argsd): + Graph.__init__(self, *argsl, **argsd) + + self.shape_files = list() + self.formats = [ + 'canon', 'cmap', 'cmapx', + 'cmapx_np', 'dia', 'dot', + 'fig', 'gd', 'gd2', 'gif', + 'hpgl', 'imap', 'imap_np', 'ismap', + 'jpe', 'jpeg', 'jpg', 'mif', + 'mp', 'pcl', 'pdf', 'pic', 'plain', + 'plain-ext', 'png', 'ps', 'ps2', + 'svg', 'svgz', 'vml', 'vmlz', + 'vrml', 'vtx', 'wbmp', 'xdot', 'xlib'] + + self.prog = 'dot' + + # Automatically creates all + # the methods enabling the creation + # of output in any of the supported formats. + for frmt in self.formats: + def new_method( + f=frmt, prog=self.prog, + encoding=None): + """Refer to docstring of method `create`.""" + return self.create( + format=f, prog=prog, encoding=encoding) + name = 'create_{fmt}'.format(fmt=frmt) + self.__setattr__(name, new_method) + + for frmt in self.formats+['raw']: + def new_method( + path, f=frmt, prog=self.prog, + encoding=None): + """Refer to docstring of method `write.`""" + self.write( + path, format=f, prog=prog, + encoding=encoding) + name = 'write_{fmt}'.format(fmt=frmt) + self.__setattr__(name, new_method) + + def __getstate__(self): + + dict = copy.copy(self.obj_dict) + + return dict + + def __setstate__(self, state): + + self.obj_dict = state + + + def set_shape_files(self, file_paths): + """Add the paths of the required image files. + + If the graph needs graphic objects to + be used as shapes or otherwise + those need to be in the same folder as + the graph is going to be rendered + from. Alternatively the absolute path to + the files can be specified when + including the graphics in the graph. + + The files in the location pointed to by + the path(s) specified as arguments + to this method will be copied to + the same temporary location where the + graph is going to be rendered. + """ + + if isinstance( file_paths, str_type): + self.shape_files.append( file_paths ) + + if isinstance( file_paths, (list, tuple) ): + self.shape_files.extend( file_paths ) + + + def set_prog(self, prog): + """Sets the default program. + + Sets the default program in charge of processing + the dot file into a graph. + """ + self.prog = prog + + + def write(self, path, prog=None, format='raw', encoding=None): + """Writes a graph to a file. + + Given a filename 'path' it will open/create and truncate + such file and write on it a representation of the graph + defined by the dot object in the format specified by + 'format' and using the encoding specified by `encoding` for text. + The format 'raw' is used to dump the string representation + of the Dot object, without further processing. + The output can be processed by any of graphviz tools, defined + in 'prog', which defaults to 'dot' + Returns True or False according to the success of the write + operation. + + There's also the preferred possibility of using: + + write_'format'(path, prog='program') + + which are automatically defined for all the supported formats. + [write_ps(), write_gif(), write_dia(), ...] + + The encoding is passed to `open` [1]. + + [1] https://docs.python.org/3/library/functions.html#open + """ + if prog is None: + prog = self.prog + if format == 'raw': + s = self.to_string() + if not PY3: + s = unicode(s) + with io.open(path, mode='wt', encoding=encoding) as f: + f.write(s) + else: + s = self.create(prog, format, encoding=encoding) + with io.open(path, mode='wb') as f: + f.write(s) + return True + + def create(self, prog=None, format='ps', encoding=None): + """Creates and returns a binary image for the graph. + + create will write the graph to a temporary dot file in the + encoding specified by `encoding` and process it with the + program given by 'prog' (which defaults to 'twopi'), reading + the binary image output and return it as: + + - `str` of bytes in Python 2 + - `bytes` in Python 3 + + There's also the preferred possibility of using: + + create_'format'(prog='program') + + which are automatically defined for all the supported formats, + for example: + + - `create_ps()` + - `create_gif()` + - `create_dia()` + + If 'prog' is a list, instead of a string, + then the fist item is expected to be the program name, + followed by any optional command-line arguments for it: + + [ 'twopi', '-Tdot', '-s10' ] + + + @param prog: either: + + - name of GraphViz executable that + can be found in the `$PATH`, or + + - absolute path to GraphViz executable. + + If you have added GraphViz to the `$PATH` and + use its executables as installed + (without renaming any of them) + then their names are: + + - `'dot'` + - `'twopi'` + - `'neato'` + - `'circo'` + - `'fdp'` + - `'sfdp'` + + On Windows, these have the notorious ".exe" extension that, + only for the above strings, will be added automatically. + + The `$PATH` is inherited from `os.env['PATH']` and + passed to `subprocess.Popen` using the `env` argument. + + If you haven't added GraphViz to your `$PATH` on Windows, + then you may want to give the absolute path to the + executable (for example, to `dot.exe`) in `prog`. + """ + + if prog is None: + prog = self.prog + + assert prog is not None + + if isinstance(prog, (list, tuple)): + prog, args = prog[0], prog[1:] + else: + args = [] + + # temp file + tmp_fd, tmp_name = tempfile.mkstemp() + os.close(tmp_fd) + self.write(tmp_name, encoding=encoding) + tmp_dir = os.path.dirname(tmp_name) + + # For each of the image files... + for img in self.shape_files: + # Get its data + f = open(img, 'rb') + f_data = f.read() + f.close() + # And copy it under a file with the same name in + # the temporary directory + f = open(os.path.join(tmp_dir, os.path.basename(img)), 'wb') + f.write(f_data) + f.close() + + arguments = ['-T{}'.format(format), ] + args + [tmp_name] + + try: + stdout_data, stderr_data, process = call_graphviz( + program=prog, + arguments=arguments, + working_dir=tmp_dir, + ) + except OSError as e: + if e.errno == errno.ENOENT: + args = list(e.args) + args[1] = '"{prog}" not found in path.'.format( + prog=prog) + raise OSError(*args) + else: + raise + + # clean file litter + for img in self.shape_files: + os.unlink(os.path.join(tmp_dir, os.path.basename(img))) + + os.unlink(tmp_name) + + if process.returncode != 0: + message = ( + '"{prog}" with args {arguments} returned code: {code}\n\n' + 'stdout, stderr:\n {out}\n{err}\n' + ).format( + prog=prog, + arguments=arguments, + code=process.returncode, + out=stdout_data, + err=stderr_data, + ) + print(message) + + assert process.returncode == 0, ( + '"{prog}" with args {arguments} returned code: {code}'.format( + prog=prog, + arguments=arguments, + code=process.returncode, + ) + ) + + return stdout_data