/usr/lib/python3/dist-packages/prov/graph.py is in python3-prov 1.5.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | from __future__ import (absolute_import, division, print_function,
unicode_literals)
__author__ = 'Trung Dong Huynh'
__email__ = 'trungdong@donggiang.com'
import networkx as nx
from prov.model import *
INFERRED_ELEMENT_CLASS = {
PROV_ATTR_ENTITY: ProvEntity,
PROV_ATTR_ACTIVITY: ProvActivity,
PROV_ATTR_AGENT: ProvAgent,
PROV_ATTR_TRIGGER: ProvEntity,
PROV_ATTR_GENERATED_ENTITY: ProvEntity,
PROV_ATTR_USED_ENTITY: ProvEntity,
PROV_ATTR_DELEGATE: ProvAgent,
PROV_ATTR_RESPONSIBLE: ProvAgent,
PROV_ATTR_SPECIFIC_ENTITY: ProvEntity,
PROV_ATTR_GENERAL_ENTITY: ProvEntity,
PROV_ATTR_ALTERNATE1: ProvEntity,
PROV_ATTR_ALTERNATE2: ProvEntity,
PROV_ATTR_COLLECTION: ProvEntity,
PROV_ATTR_INFORMED: ProvActivity,
PROV_ATTR_INFORMANT: ProvActivity
}
def prov_to_graph(prov_document):
""" Convert a :class:`~prov.model.ProvDocument` to a `MultiDiGraph
<http://networkx.github.io/documentation/latest/reference/classes.multidigraph.html>`_
instance of the `NetworkX <https://networkx.github.io/>`_ library.
:param prov_document: The :class:`~prov.model.ProvDocument` instance to convert.
"""
g = nx.MultiDiGraph()
unified = prov_document.unified()
node_map = dict()
for element in unified.get_records(ProvElement):
g.add_node(element)
node_map[element.identifier] = element
for relation in unified.get_records(ProvRelation):
# taking the first two elements of a relation
attr_pair_1, attr_pair_2 = relation.formal_attributes[:2]
# only need the QualifiedName (i.e. the value of the attribute)
qn1, qn2 = attr_pair_1[1], attr_pair_2[1]
if qn1 and qn2: # only proceed if both ends of the relation exist
try:
if qn1 not in node_map:
node_map[qn1] = \
INFERRED_ELEMENT_CLASS[attr_pair_1[0]](None, qn1)
if qn2 not in node_map:
node_map[qn2] = \
INFERRED_ELEMENT_CLASS[attr_pair_2[0]](None, qn2)
except KeyError:
# Unsupported attribute; cannot infer the type of the element
continue # skipping this relation
g.add_edge(node_map[qn1], node_map[qn2], relation=relation)
return g
def graph_to_prov(g):
""" Convert a `MultiDiGraph <http://networkx.github.io/documentation/latest/reference/classes.multidigraph.html>`_
back to a :class:`~prov.model.ProvDocument`.
:param g: The graph instance to convert.
"""
prov_doc = ProvDocument()
for n in g.nodes_iter():
if isinstance(n, ProvRecord) and n.bundle is not None:
prov_doc.add_record(n)
for _, _, edge_data in g.edges_iter(data=True):
try:
relation = edge_data['relation']
if isinstance(relation, ProvRecord):
prov_doc.add_record(relation)
except KeyError:
pass
return prov_doc
|