Source code for fastr.core.provenance

from prov.model import ProvDocument
import fastr


[docs]class Provenance(object): """ The Provenance object keeps track of everything that happens to a data object. """
[docs] def __init__(self, parent, host=None): self.parent = parent if host is not None: self.host = host.rstrip('/') else: self.host = fastr.config.web_url() self.namespaces = {} self.document = ProvDocument() # Define default namespaces self.fastr = self._add_namespace('fastr') self.tool = self._add_namespace('tool') self.node = self._add_namespace('node') self.job = self._add_namespace('job') self.data = self._add_namespace('data') self.worker = self._add_namespace('worker') self.network = self._add_namespace('network') self.fastr_info = self._add_namespace('fastrinfo') # Add fastr agent self.fastr_agent = self.agent(self.fastr[fastr.version.version])
def _add_namespace(self, name, parent=None, url=None): if parent is None: host = self.host else: if parent in self.namespaces.keys(): host = self.namespaces[parent].uri else: return None if url is None: self.namespaces[name] = self.document.add_namespace(name, "{}/{}/".format(host, name)) else: self.namespaces[name] = self.document.add_namespace(name, "{}/{}".format(host, url)) return self.namespaces[name]
[docs] def agent(self, identifier, other_attributes=None): return self.document.agent(identifier, other_attributes)
[docs] def activity(self, identifier, start_time=None, end_time=None, other_attributes=None): return self.document.activity(identifier, start_time, end_time, other_attributes)
[docs] def entity(self, identifier, other_attributes=None): return self.document.entity(identifier, other_attributes)