Mercurial > repos > guerler > springsuite
diff planemo/lib/python3.7/site-packages/prov/graph.py @ 1:56ad4e20f292 draft
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author | guerler |
---|---|
date | Fri, 31 Jul 2020 00:32:28 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/planemo/lib/python3.7/site-packages/prov/graph.py Fri Jul 31 00:32:28 2020 -0400 @@ -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__ = 'trungdong@donggiang.com' + + +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