ore.models.graph.Graph.from_graphml() - Code Metrics - troeger/fuzzed - Measure and Improve Code Quality continuously with Scrutinizer

ore.models.graph.Graph.from_graphml() D
last analyzed 2020-09-24 10:09 UTC

↳ Parent: ore.models.graph

Complexity

Conditions

Size

Total Lines	78
Code Lines	60

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	60
dl	0
loc	78
rs	4.2
c	0
b	0
f	0
cc	13
nop	2

How to fix Long Method Complexity

import json
import notations

from django.contrib.auth.models import User
from django.db import models
from django.db.models.aggregates import Max

import pyxb.utils.domutils
try:
    from .xml_fuzztree import FuzzTree as XmlFuzzTree, Namespace as XmlFuzzTreeNamespace, CreateFromDocument as fuzzTreeFromXml
    from .xml_faulttree import FaultTree as XmlFaultTree, Namespace as XmlFaultTreeNamespace, CreateFromDocument as faultTreeFromXml
    from .node_rendering import tikz_shapes
except Exception:
    print "ERROR: Perform a build process first."
    exit(-1)
from defusedxml.ElementTree import fromstring as parseXml

from .project import Project

import logging
logger = logging.getLogger('ore')


class Graph(models.Model):

    """
    Class: Graph

    This class models a generic graph that is suitable for any diagram notation. It basically serves a container for
    its contained nodes and edges. Additionally, it provides functionality for serializing it.

    Fields:
     {str}            kind         - unique identifier that indicates the graph's notation (e.g. fuzztree). Must be an
                                     element of the set of available notations (See also: <notations>)
     {str}            name         - the name of the graph
     {User}           owner        - a link to the owner of the graph
     {Project}        project      - the project corresponding to the graph
     {const datetime} created      - timestamp of the moment of graph creation (default: now)
     {bool}           deleted      - flag indicating whether this graph was deleted or not. Simplifies restoration of the
                                     graph if needed by toggling this member (default: False)
     {JSON}           graph_issues -
    """
    class Meta:
        app_label = 'ore'

    kind = models.CharField(max_length=127, choices=notations.choices)
    name = models.CharField(max_length=255)
    owner = models.ForeignKey(User, related_name='graphs')
    project = models.ForeignKey(Project, related_name='graphs')
    created = models.DateTimeField(auto_now_add=True, editable=False)
    modified = models.DateTimeField(auto_now=True)
    deleted = models.BooleanField(default=False)
    read_only = models.BooleanField(default=False)

    def __unicode__(self):
        return unicode(

            '%s%s' % ('[DELETED] ' if self.deleted else '', self.name))

    def ensure_default_nodes(self):
        """
            Add nodes that are contained in this kind of graph by default,
            in case they are missing.
        """
        notation = notations.by_kind[self.kind]
        if 'defaults' in notation:
            from .node import Node
            for index, default_node in enumerate(
                    notation['defaults']['nodes']):
                default_node.update({'properties': {}})
                # use index as node client ID
                # this is unique since all other client IDs are time stamps
                node, created = Node.objects.get_or_create(kind=default_node['kind'],
                                                           graph=self, deleted=False,
                                                           defaults={'client_id': int(index),
                                                                     'x': default_node['x'],
                                                                     'y': default_node['y']
                                                                     }
                                                           )
                if created:
                    node.save()

    def top_node(self):
        """
        Method: top_node

        Return the top node of this graph, if applicable for the given type.

        Returns:
         {Node} instance
        """
        if self.kind in {'faulttree', 'fuzztree'}:
            return self.nodes.all().get(kind='topEvent')
        else:
            return None

    def to_json(self, use_value_dict=False):
        """
        Method: to_json

        Serializes the graph into a JSON object.

        Returns:
         {dict} the graph in JSON representation
        """
        return json.dumps(self.to_dict(use_value_dict))

    def to_dict(self, use_value_dict=False):
        """
        Method: to_dict

        Encodes the whole graph as dictionary.

        Returns:
         {dict} the graph as dictionary
        """
        node_set = self.nodes.filter(deleted=False)
        edge_set = self.edges.filter(deleted=False)
        group_set = self.groups.filter(deleted=False)
        nodes = [node.to_dict(use_value_dict) for node in node_set]
        edges = [edge.to_dict(use_value_dict) for edge in edge_set]
        groups = [group.to_dict(use_value_dict) for group in group_set]

        node_seed = self.nodes.aggregate(Max('client_id'))['client_id__max']
        edge_seed = self.edges.aggregate(Max('client_id'))['client_id__max']
        group_seed = self.groups.aggregate(Max('client_id'))['client_id__max']

        return {
            'id': self.pk,
            'seed': max(node_seed, edge_seed, group_seed),
            'name': self.name,
            'type': self.kind,
            'readOnly': self.read_only,
            'nodes': nodes,
            'edges': edges,
            'nodeGroups': groups
        }

    def to_bool_term(self):
        root = self.nodes.get(kind__exact='topEvent')
        return root.to_bool_term()

    def to_graphml(self):
        graphKindData = '        <data key="kind">%s</data>\n' % (self.kind)
        if self.kind in {'faulttree', 'fuzztree'}:
            missionData = '        <data key="missionTime">%d</data>\n' % (
                self.top_node().get_property('missionTime'),)
        else:
            missionData = ''

        return ''.join([
            '<?xml version="1.0" encoding="utf-8"?>\n'
            '<graphml xmlns="http://graphml.graphdrawing.org/xmlns"\n'
            '         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"\n'
            '         xsi:schemaLocation="http://graphml.graphdrawing.org/xmlns\n'
            '                             http://graphml.graphdrawing.org/xmlns/1.0/graphml.xsd">\n'
            '    <graph id="graph" edgedefault="directed">\n',
            notations.graphml_keys[self.kind],
            '\n',
            graphKindData,
            missionData] +
            [node.to_graphml() for node in self.nodes.filter(deleted=False)] +
            [edge.to_graphml() for edge in self.edges.filter(deleted=False)] +

            ['    </graph>\n'
             '</graphml>\n'
             ])

    def to_tikz(self):
        """
        Method: to_tikz
            Translates the graph into a LaTex TIKZ representation.

        Returns:
            {string} The TIKZ representation of the graph
        """
        # Latex preambel
        result = """
\\documentclass{article}
\\usepackage[landscape, top=1in, bottom=1in, left=1in, right=1in]{geometry}
\\usepackage{helvet}
\\usepackage{adjustbox}
\\renewcommand{\\familydefault}{\\sfdefault}
\\usepackage{tikz}
\\usetikzlibrary{positioning, trees, svg.path}
\\tikzset{shapeStyle/.style={inner sep=0em, outer sep=0em}}
\\tikzset{shapeStyleDashed/.style={inner sep=0em, outer sep=0em, dashed, dash pattern=on 4.2 off 1.4}}
\\tikzset{mirrorStyle/.style={fill=white,text width=50pt, below, align=center, inner sep=0.2em, outer sep=0.3em}}
\\tikzset{fork edge/.style={line width=1.4, to path={|- ([shift={(\\tikztotarget)}] +0pt,+18pt) -| (\\tikztotarget) }}}
\\begin{document}
\\pagestyle{empty}
        """
        result += tikz_shapes + """
\\begin{figure}
\\begin{adjustbox}{max size={\\textwidth}{\\textheight}}
\\begin{tikzpicture}[auto, trim left]
        """
        # Find most left node and takes it's x coordinate as start offset
        # This basically shifts the whole tree to the left border
        minx = self.nodes.aggregate(min_x=models.Min('x'))['min_x']
        # Find root node and start from there
        # Use the TOP node Y coordinate as starting point at the upper border
        # Note: (0,0) is the upper left corder in TiKZ, but the lower left in
        # the DB
        top_event = self.nodes.get(kind='topEvent')
        result += top_event.to_tikz(x_offset=-minx, y_offset=top_event.y)
#        result += top_event.to_tikz_tree()
        result += """
\\end{tikzpicture}
\\end{adjustbox}
\\end{figure}
\\end{document}
        """
        return result

    def to_xml(self, xmltype=None):
        """
        Method: to_xml
            Serializes the graph into its XML representation.

        Returns:
            {string} The XML representation of the graph
        """
        bds = pyxb.utils.domutils.BindingDOMSupport()
        if xmltype:
            kind = xmltype
        else:
            kind = self.kind
        if kind == "fuzztree":
            tree = XmlFuzzTree(name=self.name, id=self.pk)
            bds.DeclareNamespace(XmlFuzzTreeNamespace, 'fuzzTree')
        elif kind == "faulttree":
            tree = XmlFaultTree(name=self.name, id=self.pk)
            bds.DeclareNamespace(XmlFaultTreeNamespace, 'faultTree')
        else:
            raise ValueError('No XML support for this graph type.')

        # Find root node and start from there
        top_event = self.nodes.get(kind='topEvent')
        tree.topEvent = top_event.to_xml(kind)

        dom = tree.toDOM(bds)
        return dom.toprettyxml()

    def from_xml(self, xml):
        ''' Fill this graph with the information gathered from the XML.'''
        from .node import Node
        if self.kind == "fuzztree":
            tree = fuzzTreeFromXml(xml)
        elif self.kind == "faulttree":
            tree = faultTreeFromXml(xml)
        else:
            raise ValueError('No XML support for this graph type.')

        self.name = tree.name
        self.save()
        top = Node(graph=self)
        top.load_xml(tree.topEvent)

    def from_graphml(self, graphml):
        '''
            Parses the given GraphML with the DefusedXML library, for better security.
        '''
        from .node import Node, new_client_id
        from .edge import Edge
        dom = parseXml(graphml)
        graph = dom.find('{http://graphml.graphdrawing.org/xmlns}graph')
        if graph is None:
            raise Exception(
                'Could not find <graph> element in the input data.')
        if graph.get('edgedefault') != 'directed':
            raise Exception(
                'Only GraphML documents with directed edges are supported.')
        # Determine graph type, in order to check for the right format rules
        graph_kind_element = graph.find(
            "{http://graphml.graphdrawing.org/xmlns}data[@key='kind']")
        if graph_kind_element is None:
            raise Exception(
                'Missing <data> element for graph kind declaration.')
        graph_kind = graph_kind_element.text
        if graph_kind not in notations.graphml_node_data:
            raise Exception('Invalid graph kind declaration.')
        else:
            self.kind = graph_kind
        # Save graph object to get a valid, which is needed in the node refs
        self.save()
        # go through all GraphML nodes and create them as model nodes for this
        # graph
        id_map = {}           # GraphML node ID's to node model PK's
        for gml_node in graph.iter(
                '{http://graphml.graphdrawing.org/xmlns}node'):
            gml_node_id = gml_node.get('id')
            node = Node(graph=self, client_id=int(gml_node_id))
            node.save()     # get a valid pk
            id_map[gml_node_id] = node.pk
            # Scan all data elements for the node
            for gml_node_data in gml_node.iter(
                    '{http://graphml.graphdrawing.org/xmlns}data'):
                name = gml_node_data.get('key')
                value = gml_node_data.text
                if name not in notations.graphml_node_data[graph_kind]:
                    raise Exception("Invalid graph node element '%s'" % name)
                # set according node properties
                if name == 'kind':
                    node.kind = value
                    node.save()
                else:
                    node.set_attr(name, value)
            # logger.debug("New node from GraphML import: "+str(node.to_dict()))
        # Graph properties belong to the TOP node
        # GraphML files without graph <data> properties may refer to a graph type that has no
        # TOP node concept
        for data in graph.findall(
                '{http://graphml.graphdrawing.org/xmlns}data'):
            name = data.get('key')
            if name != 'kind':          # already handled above
                if name not in notations.graphml_graph_data[graph_kind]:
                    raise Exception("Invalid graph data element '%s'" % name)
                logger.debug(
                    "Setting attribute %s to %s on top node" %
                    (name, data.text))
                self.top_node().set_attr(
                    name,
                    data.text)   # Fetch TOP node here, not existent in RBD's

        # go through all GraphML edges and create them as model edges for this
        # graph
        for gml_edge in graph.iter(
                '{http://graphml.graphdrawing.org/xmlns}edge'):
            source = Node.objects.get(pk=id_map[gml_edge.get('source')])
            target = Node.objects.get(pk=id_map[gml_edge.get('target')])
            edge = Edge(
                graph=self,
                source=source,
                target=target,
                client_id=new_client_id())
            edge.save()

    def copy_values(self, other):
        # copy all nodes, groups and their properties
        node_cache = {}

        for node in other.nodes.all():
            # first cache the old node's properties
            properties = node.properties.all()

            # create node copy by overwriting the ID field
            old_id = node.pk
            node.pk = None
            node.graph = self
            node.save()
            node_cache[old_id] = node

            # now save the property objects for the new node
            for prop in properties:
                prop.pk = None
                prop.node = node
                prop.save()

        for edge in other.edges.all():
            properties = edge.properties.all()
            edge.pk = None
            edge.source = node_cache[edge.source.pk]
            edge.target = node_cache[edge.target.pk]
            edge.graph = self
            edge.save()

            # now save the property objects for the new edge
            for prop in properties:
                prop.pk = None
                prop.edge = edge
                prop.save()

        from .node_group import NodeGroup
        for group in other.groups.all():
            # create group copy by overwriting the ID field
            newgroup = NodeGroup(graph=self)
            newgroup.save()         # prepare M2M field
            for node in group.nodes.all():
                newgroup.nodes.add(node_cache[node.pk])
            newgroup.save()

            # now save the property objects for the new group
            for prop in group.properties.all():
                prop.pk = None
                prop.node_group = newgroup
                prop.save()

        self.read_only = other.read_only
        self.save()

    def delete_configurations(self):
        """
            Deletes all informations about configurations of this graph.
        """
        self.configurations.all().delete()

    def delete_results(self, kind):
        '''
            Deletes all graph analysis results of a particular kind.
        '''
        old_results = self.results.filter(kind=kind).all()
        logger.debug(
            "Deleting %u old results of kind %s" %
            (len(old_results), kind))
        old_results.delete()

    def same_as(self, graph):
        '''
            Checks if this graph is equal to the given one in terms of nodes and their properties.
            This is a very expensive operation that is only intended for testing purposes.
            Comparing the edges is not easily possible, since the source / target ID's in the edge
            instances would need to be mapped between original and copy.
        '''

        def error_reporting():
            logger.debug("Graphs are differing:")
            text1 = self.to_graphml()
            text2 = graph.to_graphml()
            import difflib
            difflines = difflib.unified_diff(text1, text2)
            logger.debug('\n'.join(difflines))

        for my_node in self.nodes.all().filter(deleted=False):
            found_match = False
            for their_node in graph.nodes.all().filter(deleted=False):
                logger.debug(
                    "Checking our %s (%u) against %s (%u)" %
                    (str(my_node), my_node.pk, str(their_node), their_node.pk))
                if my_node.same_as(their_node):
                    logger.debug("Match")
                    found_match = True
                    break
            if not found_match:
                logger.debug(
                    "Couldn't find a match for node %s" %
                    (str(my_node)))
                error_reporting()
                return False

        for my_group in self.groups.all().filter(deleted=False):
            found_match = False
            for their_group in graph.groups.all().filter(deleted=False):
                if my_group.same_as(their_group):
                    found_match = True
                    break
            if not found_match:
                error_reporting()
                return False

        return True


1			import json
2			import notations
3
4			from django.contrib.auth.models import User
5			from django.db import models
6			from django.db.models.aggregates import Max
7
8			import pyxb.utils.domutils
9			try:
10			from .xml_fuzztree import FuzzTree as XmlFuzzTree, Namespace as XmlFuzzTreeNamespace, CreateFromDocument as fuzzTreeFromXml
11			from .xml_faulttree import FaultTree as XmlFaultTree, Namespace as XmlFaultTreeNamespace, CreateFromDocument as faultTreeFromXml
12			from .node_rendering import tikz_shapes
13			except Exception:
14			print "ERROR: Perform a build process first."
15			exit(-1)
16			from defusedxml.ElementTree import fromstring as parseXml
17
18			from .project import Project
19
20			import logging
21			logger = logging.getLogger('ore')
22
23
24			class Graph(models.Model):
25
26			"""
27			Class: Graph
28
29			This class models a generic graph that is suitable for any diagram notation. It basically serves a container for
30			its contained nodes and edges. Additionally, it provides functionality for serializing it.
31
32			Fields:
33			{str} kind - unique identifier that indicates the graph's notation (e.g. fuzztree). Must be an
34			element of the set of available notations (See also: <notations>)
35			{str} name - the name of the graph
36			{User} owner - a link to the owner of the graph
37			{Project} project - the project corresponding to the graph
38			{const datetime} created - timestamp of the moment of graph creation (default: now)
39			{bool} deleted - flag indicating whether this graph was deleted or not. Simplifies restoration of the
40			graph if needed by toggling this member (default: False)
41			{JSON} graph_issues -
42			"""
43			class Meta:
44			app_label = 'ore'
45
46			kind = models.CharField(max_length=127, choices=notations.choices)
47			name = models.CharField(max_length=255)
48			owner = models.ForeignKey(User, related_name='graphs')
49			project = models.ForeignKey(Project, related_name='graphs')
50			created = models.DateTimeField(auto_now_add=True, editable=False)
51			modified = models.DateTimeField(auto_now=True)
52			deleted = models.BooleanField(default=False)
53			read_only = models.BooleanField(default=False)
54
55			def __unicode__(self):
56			return unicode(
			0 ignored issues – show Comprehensibility Best Practice introduced 2018-03-07 23:29 UTC by Report Bug Copy Issue Report The variable `unicode` does not seem to be defined. Loading history...
57			'%s%s' % ('[DELETED] ' if self.deleted else '', self.name))
58
59			def ensure_default_nodes(self):
60			"""
61			Add nodes that are contained in this kind of graph by default,
62			in case they are missing.
63			"""
64			notation = notations.by_kind[self.kind]
65			if 'defaults' in notation:
66			from .node import Node
67			for index, default_node in enumerate(
68			notation['defaults']['nodes']):
69			default_node.update({'properties': {}})
70			# use index as node client ID
71			# this is unique since all other client IDs are time stamps
72			node, created = Node.objects.get_or_create(kind=default_node['kind'],
73			graph=self, deleted=False,
74			defaults={'client_id': int(index),
75			'x': default_node['x'],
76			'y': default_node['y']
77			}
78			)
79			if created:
80			node.save()
81
82			def top_node(self):
83			"""
84			Method: top_node
85
86			Return the top node of this graph, if applicable for the given type.
87
88			Returns:
89			{Node} instance
90			"""
91			if self.kind in {'faulttree', 'fuzztree'}:
92			return self.nodes.all().get(kind='topEvent')
93			else:
94			return None
95
96			def to_json(self, use_value_dict=False):
97			"""
98			Method: to_json
99
100			Serializes the graph into a JSON object.
101
102			Returns:
103			{dict} the graph in JSON representation
104			"""
105			return json.dumps(self.to_dict(use_value_dict))
106
107			def to_dict(self, use_value_dict=False):
108			"""
109			Method: to_dict
110
111			Encodes the whole graph as dictionary.
112
113			Returns:
114			{dict} the graph as dictionary
115			"""
116			node_set = self.nodes.filter(deleted=False)
117			edge_set = self.edges.filter(deleted=False)
118			group_set = self.groups.filter(deleted=False)
119			nodes = [node.to_dict(use_value_dict) for node in node_set]
120			edges = [edge.to_dict(use_value_dict) for edge in edge_set]
121			groups = [group.to_dict(use_value_dict) for group in group_set]
122
123			node_seed = self.nodes.aggregate(Max('client_id'))['client_id__max']
124			edge_seed = self.edges.aggregate(Max('client_id'))['client_id__max']
125			group_seed = self.groups.aggregate(Max('client_id'))['client_id__max']
126
127			return {
128			'id': self.pk,
129			'seed': max(node_seed, edge_seed, group_seed),
130			'name': self.name,
131			'type': self.kind,
132			'readOnly': self.read_only,
133			'nodes': nodes,
134			'edges': edges,
135			'nodeGroups': groups
136			}
137
138			def to_bool_term(self):
139			root = self.nodes.get(kind__exact='topEvent')
140			return root.to_bool_term()
141
142			def to_graphml(self):
143			graphKindData = ' <data key="kind">%s</data>\n' % (self.kind)
144			if self.kind in {'faulttree', 'fuzztree'}:
145			missionData = ' <data key="missionTime">%d</data>\n' % (
146			self.top_node().get_property('missionTime'),)
147			else:
148			missionData = ''
149
150			return ''.join([
151			'<?xml version="1.0" encoding="utf-8"?>\n'
152			'<graphml xmlns="http://graphml.graphdrawing.org/xmlns"\n'
153			' xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"\n'
154			' xsi:schemaLocation="http://graphml.graphdrawing.org/xmlns\n'
155			' http://graphml.graphdrawing.org/xmlns/1.0/graphml.xsd">\n'
156			' <graph id="graph" edgedefault="directed">\n',
157			notations.graphml_keys[self.kind],
158			'\n',
159			graphKindData,
160			missionData] +
161			[node.to_graphml() for node in self.nodes.filter(deleted=False)] +
162			[edge.to_graphml() for edge in self.edges.filter(deleted=False)] +
163
164			[' </graph>\n'
165			'</graphml>\n'
166			])
167
168			def to_tikz(self):
169			"""
170			Method: to_tikz
171			Translates the graph into a LaTex TIKZ representation.
172
173			Returns:
174			{string} The TIKZ representation of the graph
175			"""
176			# Latex preambel
177			result = """
178			\\documentclass{article}
179			\\usepackage[landscape, top=1in, bottom=1in, left=1in, right=1in]{geometry}
180			\\usepackage{helvet}
181			\\usepackage{adjustbox}
182			\\renewcommand{\\familydefault}{\\sfdefault}
183			\\usepackage{tikz}
184			\\usetikzlibrary{positioning, trees, svg.path}
185			\\tikzset{shapeStyle/.style={inner sep=0em, outer sep=0em}}
186			\\tikzset{shapeStyleDashed/.style={inner sep=0em, outer sep=0em, dashed, dash pattern=on 4.2 off 1.4}}
187			\\tikzset{mirrorStyle/.style={fill=white,text width=50pt, below, align=center, inner sep=0.2em, outer sep=0.3em}}
188			\\tikzset{fork edge/.style={line width=1.4, to path={\|- ([shift={(\\tikztotarget)}] +0pt,+18pt) -\| (\\tikztotarget) }}}
189			\\begin{document}
190			\\pagestyle{empty}
191			"""
192			result += tikz_shapes + """
193			\\begin{figure}
194			\\begin{adjustbox}{max size={\\textwidth}{\\textheight}}
195			\\begin{tikzpicture}[auto, trim left]
196			"""
197			# Find most left node and takes it's x coordinate as start offset
198			# This basically shifts the whole tree to the left border
199			minx = self.nodes.aggregate(min_x=models.Min('x'))['min_x']
200			# Find root node and start from there
201			# Use the TOP node Y coordinate as starting point at the upper border
202			# Note: (0,0) is the upper left corder in TiKZ, but the lower left in
203			# the DB
204			top_event = self.nodes.get(kind='topEvent')
205			result += top_event.to_tikz(x_offset=-minx, y_offset=top_event.y)
206			# result += top_event.to_tikz_tree()
207			result += """
208			\\end{tikzpicture}
209			\\end{adjustbox}
210			\\end{figure}
211			\\end{document}
212			"""
213			return result
214
215			def to_xml(self, xmltype=None):
216			"""
217			Method: to_xml
218			Serializes the graph into its XML representation.
219
220			Returns:
221			{string} The XML representation of the graph
222			"""
223			bds = pyxb.utils.domutils.BindingDOMSupport()
224			if xmltype:
225			kind = xmltype
226			else:
227			kind = self.kind
228			if kind == "fuzztree":
229			tree = XmlFuzzTree(name=self.name, id=self.pk)
230			bds.DeclareNamespace(XmlFuzzTreeNamespace, 'fuzzTree')
231			elif kind == "faulttree":
232			tree = XmlFaultTree(name=self.name, id=self.pk)
233			bds.DeclareNamespace(XmlFaultTreeNamespace, 'faultTree')
234			else:
235			raise ValueError('No XML support for this graph type.')
236
237			# Find root node and start from there
238			top_event = self.nodes.get(kind='topEvent')
239			tree.topEvent = top_event.to_xml(kind)
240
241			dom = tree.toDOM(bds)
242			return dom.toprettyxml()
243
244			def from_xml(self, xml):
245			''' Fill this graph with the information gathered from the XML.'''
246			from .node import Node
247			if self.kind == "fuzztree":
248			tree = fuzzTreeFromXml(xml)
249			elif self.kind == "faulttree":
250			tree = faultTreeFromXml(xml)
251			else:
252			raise ValueError('No XML support for this graph type.')
253
254			self.name = tree.name
255			self.save()
256			top = Node(graph=self)
257			top.load_xml(tree.topEvent)
258
259			def from_graphml(self, graphml):
260			'''
261			Parses the given GraphML with the DefusedXML library, for better security.
262			'''
263			from .node import Node, new_client_id
264			from .edge import Edge
265			dom = parseXml(graphml)
266			graph = dom.find('{http://graphml.graphdrawing.org/xmlns}graph')
267			if graph is None:
268			raise Exception(
269			'Could not find <graph> element in the input data.')
270			if graph.get('edgedefault') != 'directed':
271			raise Exception(
272			'Only GraphML documents with directed edges are supported.')
273			# Determine graph type, in order to check for the right format rules
274			graph_kind_element = graph.find(
275			"{http://graphml.graphdrawing.org/xmlns}data[@key='kind']")
276			if graph_kind_element is None:
277			raise Exception(
278			'Missing <data> element for graph kind declaration.')
279			graph_kind = graph_kind_element.text
280			if graph_kind not in notations.graphml_node_data:
281			raise Exception('Invalid graph kind declaration.')
282			else:
283			self.kind = graph_kind
284			# Save graph object to get a valid, which is needed in the node refs
285			self.save()
286			# go through all GraphML nodes and create them as model nodes for this
287			# graph
288			id_map = {} # GraphML node ID's to node model PK's
289			for gml_node in graph.iter(
290			'{http://graphml.graphdrawing.org/xmlns}node'):
291			gml_node_id = gml_node.get('id')
292			node = Node(graph=self, client_id=int(gml_node_id))
293			node.save() # get a valid pk
294			id_map[gml_node_id] = node.pk
295			# Scan all data elements for the node
296			for gml_node_data in gml_node.iter(
297			'{http://graphml.graphdrawing.org/xmlns}data'):
298			name = gml_node_data.get('key')
299			value = gml_node_data.text
300			if name not in notations.graphml_node_data[graph_kind]:
301			raise Exception("Invalid graph node element '%s'" % name)
302			# set according node properties
303			if name == 'kind':
304			node.kind = value
305			node.save()
306			else:
307			node.set_attr(name, value)
308			# logger.debug("New node from GraphML import: "+str(node.to_dict()))
309			# Graph properties belong to the TOP node
310			# GraphML files without graph <data> properties may refer to a graph type that has no
311			# TOP node concept
312			for data in graph.findall(
313			'{http://graphml.graphdrawing.org/xmlns}data'):
314			name = data.get('key')
315			if name != 'kind': # already handled above
316			if name not in notations.graphml_graph_data[graph_kind]:
317			raise Exception("Invalid graph data element '%s'" % name)
318			logger.debug(
319			"Setting attribute %s to %s on top node" %
320			(name, data.text))
321			self.top_node().set_attr(
322			name,
323			data.text) # Fetch TOP node here, not existent in RBD's
324
325			# go through all GraphML edges and create them as model edges for this
326			# graph
327			for gml_edge in graph.iter(
328			'{http://graphml.graphdrawing.org/xmlns}edge'):
329			source = Node.objects.get(pk=id_map[gml_edge.get('source')])
330			target = Node.objects.get(pk=id_map[gml_edge.get('target')])
331			edge = Edge(
332			graph=self,
333			source=source,
334			target=target,
335			client_id=new_client_id())
336			edge.save()
337
338			def copy_values(self, other):
339			# copy all nodes, groups and their properties
340			node_cache = {}
341
342			for node in other.nodes.all():
343			# first cache the old node's properties
344			properties = node.properties.all()
345
346			# create node copy by overwriting the ID field
347			old_id = node.pk
348			node.pk = None
349			node.graph = self
350			node.save()
351			node_cache[old_id] = node
352
353			# now save the property objects for the new node
354			for prop in properties:
355			prop.pk = None
356			prop.node = node
357			prop.save()
358
359			for edge in other.edges.all():
360			properties = edge.properties.all()
361			edge.pk = None
362			edge.source = node_cache[edge.source.pk]
363			edge.target = node_cache[edge.target.pk]
364			edge.graph = self
365			edge.save()
366
367			# now save the property objects for the new edge
368			for prop in properties:
369			prop.pk = None
370			prop.edge = edge
371			prop.save()
372
373			from .node_group import NodeGroup
374			for group in other.groups.all():
375			# create group copy by overwriting the ID field
376			newgroup = NodeGroup(graph=self)
377			newgroup.save() # prepare M2M field
378			for node in group.nodes.all():
379			newgroup.nodes.add(node_cache[node.pk])
380			newgroup.save()
381
382			# now save the property objects for the new group
383			for prop in group.properties.all():
384			prop.pk = None
385			prop.node_group = newgroup
386			prop.save()
387
388			self.read_only = other.read_only
389			self.save()
390
391			def delete_configurations(self):
392			"""
393			Deletes all informations about configurations of this graph.
394			"""
395			self.configurations.all().delete()
396
397			def delete_results(self, kind):
398			'''
399			Deletes all graph analysis results of a particular kind.
400			'''
401			old_results = self.results.filter(kind=kind).all()
402			logger.debug(
403			"Deleting %u old results of kind %s" %
404			(len(old_results), kind))
405			old_results.delete()
406
407			def same_as(self, graph):
408			'''
409			Checks if this graph is equal to the given one in terms of nodes and their properties.
410			This is a very expensive operation that is only intended for testing purposes.
411			Comparing the edges is not easily possible, since the source / target ID's in the edge
412			instances would need to be mapped between original and copy.
413			'''
414
415			def error_reporting():
416			logger.debug("Graphs are differing:")
417			text1 = self.to_graphml()
418			text2 = graph.to_graphml()
419			import difflib
420			difflines = difflib.unified_diff(text1, text2)
421			logger.debug('\n'.join(difflines))
422
423			for my_node in self.nodes.all().filter(deleted=False):
424			found_match = False
425			for their_node in graph.nodes.all().filter(deleted=False):
426			logger.debug(
427			"Checking our %s (%u) against %s (%u)" %
428			(str(my_node), my_node.pk, str(their_node), their_node.pk))
429			if my_node.same_as(their_node):
430			logger.debug("Match")
431			found_match = True
432			break
433			if not found_match:
434			logger.debug(
435			"Couldn't find a match for node %s" %
436			(str(my_node)))
437			error_reporting()
438			return False
439
440			for my_group in self.groups.all().filter(deleted=False):
441			found_match = False
442			for their_group in graph.groups.all().filter(deleted=False):
443			if my_group.same_as(their_group):
444			found_match = True
445			break
446			if not found_match:
447			error_reporting()
448			return False
449
450			return True
451

troeger / fuzzed

ore.models.graph.Graph.from_graphml() D last analyzed 2020-09-24 10:09 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

ore.models.graph.Graph.from_graphml() D
last analyzed 2020-09-24 10:09 UTC