diff env/lib/python3.9/site-packages/prov/graph.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/prov/graph.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,93 @@
+from __future__ import (absolute_import, division, print_function,
+                        unicode_literals)
+
+import networkx as nx
+from prov.model import (
+    ProvDocument, ProvRecord, ProvElement, ProvEntity, ProvActivity, ProvAgent,
+    ProvRelation, PROV_ATTR_ENTITY, PROV_ATTR_ACTIVITY, PROV_ATTR_AGENT,
+    PROV_ATTR_TRIGGER, PROV_ATTR_GENERATED_ENTITY, PROV_ATTR_USED_ENTITY,
+    PROV_ATTR_DELEGATE, PROV_ATTR_RESPONSIBLE, PROV_ATTR_SPECIFIC_ENTITY,
+    PROV_ATTR_GENERAL_ENTITY, PROV_ATTR_ALTERNATE1, PROV_ATTR_ALTERNATE2,
+    PROV_ATTR_COLLECTION, PROV_ATTR_INFORMED, PROV_ATTR_INFORMANT
+)
+
+__author__ = 'Trung Dong Huynh'
+__email__ = '[e-mail]'
+
+
+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
+    <https://networkx.readthedocs.io/en/stable/reference/classes.multigraph.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
+    <https://networkx.readthedocs.io/en/stable/reference/classes.multigraph.html>`_
+    that was previously produced by :func:`prov_to_graph` 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