first part of the provenance collection draft

Author: notactuallyfinn

src/hermes/commands/harvest/base.py

@@ -12,17 +12,28 @@
 from hermes.error import HermesPluginRunError, MisconfigurationError
 from hermes.model.context_manager import HermesContext
 from hermes.model import SoftwareMetadata
+from hermes.model.provenance.ld_prov import ld_prov_list, ld_prov_node
+from hermes.model.types.ld_context import ALL_CONTEXTS
 
 
 class HermesHarvestPlugin(HermesPlugin):
     """Base plugin that does harvesting.
 
     TODO: describe the harvesting process and how this is mapped to this plugin.
     """
+    def __init__(self):
+        self.io_operations: list[tuple[dict, dict, dict]] = []
+        super().__init__()
 
     def __call__(self, command: HermesCommand) -> SoftwareMetadata:
         pass
 
+    def load():
+        pass
+
+    def write():
+        pass
+
 
 class HarvestSettings(BaseModel):
     """Generic harvesting settings."""
@@ -37,9 +48,11 @@ class HermesHarvestCommand(HermesCommand):
     settings_class = HarvestSettings
 
     def __call__(self, args: argparse.Namespace) -> None:
-        self.log.info("# Metadata harvesting")
         self.args = args
+        self.log.info("# Load provenance from old harvest or create new document.")
+        prov_doc, base_plugin = self.init_provenance_document()
 
+        self.log.info("# Metadata harvesting")
         if len(self.settings.sources) == 0:
             self.log.critical("# No harvest plugin was configured to be run and loaded.")
             raise MisconfigurationError("No harvest plugin was configured to be run and loaded.")
@@ -66,6 +79,24 @@ def __call__(self, args: argparse.Namespace) -> None:
             except Exception:
                 self.log.exception(f"### Unknown error while executing the {plugin_name} plugin, skipping it now.")
                 continue
+            self.remove_provenance_info_for_plugin(prov_doc, plugin_name)
+
+            plugin = prov_doc.add_hermes_plugin("harvest", plugin_name)
+            plugin_io_operations = plugin_func.io_operations # liste von drei Tupeln (input_file, load_function, output)
+            for plugin_io_operation in plugin_io_operations:
+                loaded_source = prov_doc.add_entity()
+                loaded_source.update(plugin_io_operation[0])
+                io_op = prov_doc.add_activity()
+                plugin_io_operation[1]["prov:wasAssociatedWith"] = [base_plugin.ref, plugin.ref]
+                plugin_io_operation[1]["prov:used"] = loaded_source.ref
+                io_op.update(plugin_io_operation[1])
+                loaded_data = prov_doc.add_entity()
+                plugin_io_operation[2].update({
+                    "prov:wasAttributedTo": plugin.ref,
+                    "prov:wasDerivedFrom": loaded_source.ref,
+                    "prov:wasGeneratedBy": io_op.ref
+                })
+                loaded_data.update(plugin_io_operations[2])
 
             self.log.info(f"### Store metadata harvested by {plugin_name} plugin")
             # store harvested data
@@ -76,3 +107,37 @@ def __call__(self, args: argparse.Namespace) -> None:
         if not harvested_any:
             self.log.critical("No harvest plugin ran successfully.")
             raise HermesPluginRunError("No harvest plugin ran successfully.")
+
+    def init_provenance_document(self) -> tuple[ld_prov_list, ld_prov_node]:
+        ctx = HermesContext()
+        ctx.prepare_step("harvest")
+        with ctx["provenance"] as cache:
+            try:
+                ld_prov_doc = ld_prov_list.from_list(cache["codemeta"], container_type="@graph", context=ALL_CONTEXTS)
+                return ld_prov_doc, ld_prov_doc.shallow_search({"schema:name": lambda doc, node: node["schema:name"][0].find("harvest base plugin") != -1})[0]
+            except KeyError:
+                pass
+        prov_doc = ld_prov_list()
+        prov_doc.init_hermes_agents()
+        return prov_doc, prov_doc.add_hermes_base_plugin("harvest")
+
+    def remove_provenance_info_for_plugin(self, prov_doc, plugin) -> None:
+        plugin = prov_doc.shallow_search({
+            "schema:name": (lambda doc, node: f"harvest plugin {plugin}" in node["schema:name"]),
+        })
+        if len(plugin) == 0:
+            return
+        # two passes are needed because the nodes are nested exactly two levels
+        related = prov_doc.shallow_search({
+            "prov:wasAssociatedWith": (lambda doc, node: plugin.ref in node["prov:wasAssociatedWith"]),
+            "prov:wasAttributedTo": (lambda doc, node: plugin.ref in node["prov:wasAttributedTo"])
+        })
+        ids = [plugin.ref, *(rel.ref for rel in related)]
+        related = prov_doc.shallow_search({
+            f"prov:{key}": (lambda doc, node: any(id in node[f"prov:{key}"] for id in ids)) for key in [
+                "wasAssociatedWith", "wasAttributedTo", "wasGeneratedBy", "used", "wasDerivedFrom", "wasInformedBy"
+            ]
+        })
+        for item in related:
+            items = prov_doc.shallow_search({"@id": (lambda doc, node: node["@id"] == item["@id"])})
+            del prov_doc[items[0].index]

src/hermes/model/provenance/ld_prov.py

@@ -0,0 +1,158 @@
+# SPDX-FileCopyrightText: 2026 German Aerospace Center (DLR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Fritzsche
+
+from typing import Optional, Union
+from typing_extensions import Self
+import uuid
+
+from hermes import utils
+from hermes.model.types import ld_dict, ld_list
+from hermes.model.types.ld_container import BASIC_TYPE, EXPANDED_JSON_LD_VALUE, JSON_LD_CONTEXT_DICT, TIME_TYPE
+from hermes.model.types.ld_context import ALL_CONTEXTS, iri_map
+
+
+class ld_prov_container:
+    def _to_python(
+        self: Self,
+        full_iri: str,
+        ld_value: Union[EXPANDED_JSON_LD_VALUE, dict[str, EXPANDED_JSON_LD_VALUE], list[str], str]
+    ) -> Union["ld_prov_node", "ld_prov_list", BASIC_TYPE, TIME_TYPE]:
+        item = super()._to_python(full_iri, ld_value)
+        if isinstance(item, ld_list):
+            return ld_prov_list(
+                data=item._data, parent=item.parent, key=item.key, index=item.index, context=item.context
+            )
+        elif isinstance(item, ld_dict):
+            return ld_prov_node(
+                data=item._data, parent=item.parent, key=item.key, index=item.index, context=item.context
+            )
+        return item
+
+
+class ld_prov_list(ld_list):
+    NODE_IRI_FORMAT = "graph://{uuid}/{index}"
+    PROV_DOC_IRI = iri_map['hermes-rt', "graph"]
+
+    def __init__(
+        self: Self,
+        *,
+        data: EXPANDED_JSON_LD_VALUE = [{"@graph": []}],
+        parent: Optional[Union[ld_dict, ld_list]] = None,
+        key: Optional[str] = PROV_DOC_IRI,
+        index: Optional[int] = None,
+        context: Optional[list[Union[str, JSON_LD_CONTEXT_DICT]]] = ALL_CONTEXTS
+    ) -> None:
+        self.id = uuid.uuid1()
+        self.node_index = 0
+        super().__init__([{"@graph": []}], parent=parent, key=key, index=index, context=context)
+
+    def __getitem__(
+        self: Self, index: Union[int, slice]
+    ) -> Union[
+        BASIC_TYPE,
+        TIME_TYPE,
+        "ld_prov_node",
+        "ld_prov_list",
+        list[Union[BASIC_TYPE, TIME_TYPE, "ld_prov_node", "ld_prov_list"]]
+    ]:
+        item = super().__getitem__(index)
+        if isinstance(item, ld_list):
+            return ld_prov_list(
+                data=item._data, parent=item.parent, key=item.key, index=item.index, context=item.context
+            )
+        elif isinstance(item, ld_dict):
+            return ld_prov_node(
+                data=item._data, parent=item.parent, key=item.key, index=item.index, context=item.context
+            )
+        return item
+
+    def next_node_iri(self) -> str:
+        self.node_index += 1
+        return self.NODE_IRI_FORMAT.format(uuid=self.id, index=self.node_index)
+
+    def add_activity(self) -> "ld_prov_node":
+        self.append({"@id": self.next_node_iri(), "@type": "prov:Activity"})
+        return self[-1]
+
+    def add_agent(self) -> "ld_prov_node":
+        self.append({"@id": self.next_node_iri(), "@type": "prov:Agent"})
+        return self[-1]
+
+    def add_entity(self) -> "ld_prov_node":
+        self.append({"@id": self.next_node_iri(), "@type": "prov:Entity"})
+        return self[-1]
+
+    def init_hermes_agents(self) -> "ld_prov_node":
+        hermes = self.add_agent()
+        hermes.update({
+            "schema:name": utils.hermes_name,
+            "schema:version": utils.hermes_version,
+            "schema:url": utils.hermes_urls,
+        })
+        hermes["@type"].append("schema:SoftwareApplication")
+        node = self.add_agent()
+        node.update({
+            "schema:name": utils.hermes_name + " cache",
+            "schema:version": utils.hermes_version,
+            "prov:actedOnBehalfOf": hermes.ref
+        })
+        node["@type"].append("schema:SoftwareApplication")
+        return node
+
+    def add_hermes_command(self, step) -> "ld_prov_node":
+        node = self.add_agent()
+        node.update({
+            "schema:name": f"{utils.hermes_name} {step} command",
+            "schema:version": utils.hermes_version,
+            "prov:actedOnBehalfOf": self.shallow_search(
+                {"schema:name": (lambda doc, node: node["schema:name"] == utils.hermes_name)}
+            )
+        })
+        node["@type"].append("schema:SoftwareApplication")
+        return node
+
+    def add_hermes_base_plugin(self, step) -> "ld_prov_node":
+        node = self.add_agent()
+        node.update({
+            "schema:name": f"{utils.hermes_name} {step} base plugin",
+            "schema:version": utils.hermes_version,
+            "prov:actedOnBehalfOf": self.shallow_search(
+                {"schema:name": (lambda doc, node: node["schema:name"] == f"{utils.hermes_name} {step} command")}
+            )
+        })
+        node["@type"].append("schema:SoftwareApplication")
+        return node
+
+    def add_hermes_plugin(self, step, name) -> "ld_prov_node":
+        node = self.add_agent()
+        # TODO: add version
+        node.update({
+            "schema:name": f"{utils.hermes_name} {step} plugin '{name}'",
+            "prov:actedOnBehalfOf": self.shallow_search(
+                {"schema:name": (lambda doc, node: node["schema:name"] == f"{utils.hermes_name} {step} base plugin")}
+            )
+        })
+        node["@type"].append("schema:SoftwareApplication")
+        return node
+
+    def shallow_search(self, query: dict) -> list["ld_prov_node"]:
+        return [
+            item for item in self for key, test in query.items() if key in item and test(self, item)
+        ]
+
+
+class ld_prov_node(ld_dict):
+    def __init__(
+        self: Self,
+        data: list[dict[str, EXPANDED_JSON_LD_VALUE]],
+        *,
+        parent: Optional[Union[ld_dict, ld_list]] = None,
+        key: Optional[str] = None,
+        index: Optional[int] = None,
+        context: Optional[list[Union[str, JSON_LD_CONTEXT_DICT]]] = ALL_CONTEXTS
+    ) -> None:
+        self.id = uuid.uuid1()
+        super().__init__(data, parent=parent, key=key, index=index, context=context)

src/hermes/model/types/ld_list.py

@@ -589,7 +589,7 @@ def from_list(
         key: Optional[str] = None,
         context: Optional[Union[str, JSON_LD_CONTEXT_DICT, list[Union[str, JSON_LD_CONTEXT_DICT]]]] = None,
         container_type: str = "@set"
-    ) -> ld_list:
+    ) -> Self:
         """
         Creates a ld_list from the given list with the given parent, key, context and container_type.\n
         Note that only container_type '@set' is valid for key '@type'.\n