From 040415b14a23e6749f985f6f3599941e3ddb3165 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Mon, 4 May 2026 16:19:06 +0200 Subject: [PATCH 01/11] first part of the provenance collection draft --- src/hermes/commands/harvest/base.py | 67 ++++++++++- src/hermes/model/provenance/ld_prov.py | 158 +++++++++++++++++++++++++ src/hermes/model/types/ld_list.py | 2 +- 3 files changed, 225 insertions(+), 2 deletions(-) create mode 100644 src/hermes/model/provenance/ld_prov.py diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 0d3d9e5f..93b23601 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -12,6 +12,8 @@ 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): @@ -19,10 +21,19 @@ class HermesHarvestPlugin(HermesPlugin): 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] diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py new file mode 100644 index 00000000..2c2a10f7 --- /dev/null +++ b/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) diff --git a/src/hermes/model/types/ld_list.py b/src/hermes/model/types/ld_list.py index 14331472..003cda82 100644 --- a/src/hermes/model/types/ld_list.py +++ b/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 From a96821eb7964dfa9c1802c3049cae1e48a3d70b1 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Mon, 4 May 2026 17:21:28 +0200 Subject: [PATCH 02/11] first draft for provenance recording of harvest command --- src/hermes/commands/harvest/base.py | 64 +++++++++++++++++++++++--- src/hermes/model/provenance/ld_prov.py | 12 ++--- 2 files changed, 64 insertions(+), 12 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 93b23601..61e2b3e7 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -79,10 +79,18 @@ 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.log.info(f"### Store metadata harvested by {plugin_name} plugin") + # store harvested data + harvested_data.write_to_cache(ctx, plugin_name) + harvested_any = True + 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) + plugin_io_operations = plugin_func.io_operations + outputs = [] + io_ops = [] for plugin_io_operation in plugin_io_operations: loaded_source = prov_doc.add_entity() loaded_source.update(plugin_io_operation[0]) @@ -97,11 +105,52 @@ def __call__(self, args: argparse.Namespace) -> None: "prov:wasGeneratedBy": io_op.ref }) loaded_data.update(plugin_io_operations[2]) + outputs.append(loaded_data.ref) + io_ops.append(io_op.ref) + + map_activity = prov_doc.add_activity() + map_activity.update({ + "prov:wasInformedBy": io_ops, + "prov:used": outputs, + "prov:wasAssociatedWith": plugin.ref + }) + data_output = prov_doc.add_entity() + data_output.update({ + "prov:wasAttributedTo": plugin.ref, + "prov:wasGeneratedBy": map_activity.ref, + "prov:wasDerivedFrom": outputs + }) + + write = prov_doc.add_activity() + write.update({ + "prov:wasAssociatedWith": [ + prov_doc.shallow_search({ + "schema:name": lambda doc, node: node["schema:name"][0].find("harvest command") != -1 + })[0].ref, + prov_doc.shallow_search({ + "schema:name": lambda doc, node: node["schema:name"][0].find(" cache") != -1 + })[0].ref, + plugin.ref + ], + "prov:used": data_output.ref, + "prov:wasInformedBy": map_activity.ref + }) + # TODO: add more info + write_output = prov_doc.add_entity() + write_output.update({ + "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + }) + write_output = prov_doc.add_entity() + write_output.update({ + "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + }) + write_output = prov_doc.add_entity() + write_output.update({ + "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + }) - self.log.info(f"### Store metadata harvested by {plugin_name} plugin") - # store harvested data - harvested_data.write_to_cache(ctx, plugin_name) - harvested_any = True + with ctx["provenance"] as cache: + cache["codemeta"] = prov_doc.ld_value ctx.finalize_step('harvest') if not harvested_any: @@ -114,11 +163,14 @@ def init_provenance_document(self) -> tuple[ld_prov_list, ld_prov_node]: 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] + 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() + prov_doc.add_hermes_command("harvest") return prov_doc, prov_doc.add_hermes_base_plugin("harvest") def remove_provenance_info_for_plugin(self, prov_doc, plugin) -> None: diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index 2c2a10f7..a8419da9 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -108,8 +108,8 @@ def add_hermes_command(self, step) -> "ld_prov_node": "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)} - ) + {"schema:name": (lambda doc, node: node["schema:name"][0] == utils.hermes_name)} + )[0].ref }) node["@type"].append("schema:SoftwareApplication") return node @@ -120,8 +120,8 @@ def add_hermes_base_plugin(self, step) -> "ld_prov_node": "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")} - ) + {"schema:name": (lambda doc, node: node["schema:name"][0] == f"{utils.hermes_name} {step} command")} + )[0].ref }) node["@type"].append("schema:SoftwareApplication") return node @@ -132,8 +132,8 @@ def add_hermes_plugin(self, step, name) -> "ld_prov_node": 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")} - ) + {"schema:name": (lambda doc, node: node["schema:name"][0] == f"{utils.hermes_name} {step} base plugin")} + )[0].ref }) node["@type"].append("schema:SoftwareApplication") return node From 4f85996d53dda0997706283c3a6aa6682b889999 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 7 May 2026 16:33:13 +0200 Subject: [PATCH 03/11] improved ld_prov_list and adjusted the collection of provenance accordingly --- src/hermes/commands/harvest/base.py | 83 ++++------ src/hermes/model/provenance/ld_prov.py | 216 ++++++++++++------------- 2 files changed, 135 insertions(+), 164 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 61e2b3e7..e6fbfc4c 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -12,7 +12,7 @@ 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.provenance.ld_prov import ld_prov_list from hermes.model.types.ld_context import ALL_CONTEXTS @@ -50,7 +50,8 @@ class HermesHarvestCommand(HermesCommand): def __call__(self, args: argparse.Namespace) -> None: self.args = args self.log.info("# Load provenance from old harvest or create new document.") - prov_doc, base_plugin = self.init_provenance_document() + prov_doc = self.init_provenance_document() + base_plugin = prov_doc.get_hermes_base_plugin("harvest") self.log.info("# Metadata harvesting") if len(self.settings.sources) == 0: @@ -92,60 +93,48 @@ def __call__(self, args: argparse.Namespace) -> None: outputs = [] io_ops = [] 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() + loaded_source = prov_doc.add_entity(data=plugin_io_operation[0]) + plugin_io_operation[1].update( + {"prov:wasAssociatedWith": [base_plugin.ref, plugin.ref], "prov:used": loaded_source.ref} + ) + io_op = prov_doc.add_activity(data=plugin_io_operation[1]) 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]) + loaded_data = prov_doc.add_entity(data=plugin_io_operations[2]) outputs.append(loaded_data.ref) io_ops.append(io_op.ref) - map_activity = prov_doc.add_activity() - map_activity.update({ + map_activity = prov_doc.add_activity(data={ "prov:wasInformedBy": io_ops, "prov:used": outputs, "prov:wasAssociatedWith": plugin.ref }) - data_output = prov_doc.add_entity() - data_output.update({ + data_output = prov_doc.add_entity(data={ "prov:wasAttributedTo": plugin.ref, "prov:wasGeneratedBy": map_activity.ref, "prov:wasDerivedFrom": outputs }) - write = prov_doc.add_activity() - write.update({ + write = prov_doc.add_activity(data={ "prov:wasAssociatedWith": [ - prov_doc.shallow_search({ - "schema:name": lambda doc, node: node["schema:name"][0].find("harvest command") != -1 - })[0].ref, - prov_doc.shallow_search({ - "schema:name": lambda doc, node: node["schema:name"][0].find(" cache") != -1 - })[0].ref, + prov_doc.get_hermes_command("harvest").ref, + prov_doc.get_hermes_cache().ref, plugin.ref ], "prov:used": data_output.ref, "prov:wasInformedBy": map_activity.ref }) # TODO: add more info - write_output = prov_doc.add_entity() - write_output.update({ + prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref }) - write_output = prov_doc.add_entity() - write_output.update({ + prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref }) - write_output = prov_doc.add_entity() - write_output.update({ + prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref }) @@ -157,39 +146,37 @@ def __call__(self, args: argparse.Namespace) -> None: 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]: + def init_provenance_document(self) -> ld_prov_list: 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] + return ld_prov_doc except KeyError: pass prov_doc = ld_prov_list() prov_doc.init_hermes_agents() - prov_doc.add_hermes_command("harvest") - 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 prov_doc + + def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> None: + plugin = prov_doc.get_hermes_plugin("harvest", plugin) + if plugin is None: 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"]) - }) + related = prov_doc.shallow_search(lambda doc, node: ( + ("prov:wasAssociatedWith" in node and plugin.ref in node["prov:wasAssociatedWith"]) or + ("prov:wasAttributedTo" in node and plugin.ref in node["prov:wasAttributedTo"]) + )) + if len(related) == 0: + del prov_doc[plugin.index] + return 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 [ + related = prov_doc.shallow_search(lambda doc, node: any( + (f"prov:{key}" in node and 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"])}) + items = prov_doc.shallow_search(lambda doc, node: ("@id" in node and node["@id"] == item["@id"])) del prov_doc[items[0].index] diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index a8419da9..df4b8788 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -6,35 +6,22 @@ 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_container import EXPANDED_JSON_LD_VALUE, JSON_LD_CONTEXT_DICT 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}" + NODE_IRI_FORMAT = "_:{type}/{index}" + HERMES_ID = f"https://doi.org/{utils.hermes_doi}" + HERMES_CACHE_ID = "_:hermes/cache" + HERMES_COMMAND_ID_FORMAT = "_:hermes/command/{step}" + HERMES_PLUGIN_ID_FORMAT = "_:hermes/plugin/{step}/{name}" + HERMES_BASE_PLUGIN_ID_FORMAT = "_:hermes/base_plugin/{step}" PROV_DOC_IRI = iri_map['hermes-rt', "graph"] + INDICES = {} def __init__( self: Self, @@ -45,114 +32,111 @@ def __init__( 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({ + def next_node_iri(self, type) -> str: + if type not in ld_prov_list.INDICES: + ld_prov_list.INDICES[type] = 0 + ld_prov_list.INDICES[type] += 1 + return self.NODE_IRI_FORMAT.format(type=type, index=ld_prov_list.INDICES[type]) + + def add_activity(self, *, data={}) -> ld_dict: + self.append(data) + activity = self[-1] + if "@type" not in data: + activity["@type"] = "prov:Activity" + else: + activity["@type"].append("prov:Activity") + if "@id" not in data: + activity["@id"] = self.next_node_iri("Activity") + return activity + + def add_agent(self, *, data={}) -> ld_dict: + self.append(data) + agent = self[-1] + if "@type" not in data: + agent["@type"] = "prov:Agent" + else: + agent["@type"].append("prov:Agent") + if "@id" not in data: + agent["@id"] = self.next_node_iri("Agent") + return agent + + def add_entity(self, *, data={}) -> ld_dict: + self.append(data) + entity = self[-1] + if "@type" not in data: + entity["@type"] = "prov:Entity" + else: + entity["@type"].append("prov:Entity") + if "@id" not in data: + entity["@id"] = self.next_node_iri("Entity") + return entity + + def init_hermes_agents(self) -> None: + hermes = self.add_agent(data={ + "@id": ld_prov_list.HERMES_ID, + "@type": "schema:SoftwareApplication", "schema:name": utils.hermes_name, "schema:version": utils.hermes_version, - "schema:url": utils.hermes_urls, + "schema:url": [*set(utils.hermes_urls.values())] }) - hermes["@type"].append("schema:SoftwareApplication") - node = self.add_agent() - node.update({ + self.add_agent(data={ + "@id": ld_prov_list.HERMES_CACHE_ID, + "@type": "schema:SoftwareApplication", "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"][0] == utils.hermes_name)} - )[0].ref - }) - 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"][0] == f"{utils.hermes_name} {step} command")} - )[0].ref - }) - node["@type"].append("schema:SoftwareApplication") - return node - - def add_hermes_plugin(self, step, name) -> "ld_prov_node": - node = self.add_agent() + for step in ["harvest", "process", "curate", "deposit", "postprocess"]: + command = self.add_agent(data={ + "@id": ld_prov_list.HERMES_COMMAND_ID_FORMAT.format(step=step), + "@type": "schema:SoftwareApplication", + "schema:name": f"{utils.hermes_name} {step} command", + "schema:version": utils.hermes_version, + "prov:actedOnBehalfOf": hermes.ref + }) + self.add_agent(data={ + "@id": ld_prov_list.HERMES_BASE_PLUGIN_ID_FORMAT.format(step=step), + "@type": "schema:SoftwareApplication", + "schema:name": f"{utils.hermes_name} {step} base plugin", + "schema:version": utils.hermes_version, + "prov:actedOnBehalfOf": command.ref + }) + + def add_hermes_plugin(self, step, name) -> ld_dict: # TODO: add version - node.update({ + node = self.add_agent(data={ + "@id": ld_prov_list.HERMES_PLUGIN_ID_FORMAT.format(step=step, name=name), + "@type": "schema:SoftwareApplication", "schema:name": f"{utils.hermes_name} {step} plugin '{name}'", - "prov:actedOnBehalfOf": self.shallow_search( - {"schema:name": (lambda doc, node: node["schema:name"][0] == f"{utils.hermes_name} {step} base plugin")} - )[0].ref + "prov:actedOnBehalfOf": self.get_hermes_base_plugin(step) }) - 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) - ] + def shallow_search(self, query) -> list[ld_dict]: + return [item for item in self if query(self, item)] + def get_hermes(self) -> ld_dict: + return self.shallow_search(lambda doc, node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_ID))[0] -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) + def get_hermes_cache(self) -> ld_dict: + return self.shallow_search(lambda doc, node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_CACHE_ID))[0] + + def get_hermes_base_plugin(self, step) -> ld_dict: + return self.shallow_search(lambda doc, node: ( + "@id" in node and node["@id"] == ld_prov_list.HERMES_BASE_PLUGIN_ID_FORMAT.format(step=step) + ))[0] + + def get_hermes_plugin(self, step, name) -> Union[ld_dict, None]: + search_result = self.shallow_search(lambda doc, node: ( + "@id" in node and node["@id"] == ld_prov_list.HERMES_PLUGIN_ID_FORMAT.format(step=step, name=name) + )) + if search_result: + return search_result[0] + return None + + def get_hermes_command(self, step) -> ld_dict: + return self.shallow_search(lambda doc, node: ( + "@id" in node and node["@id"] == ld_prov_list.HERMES_COMMAND_ID_FORMAT.format(step=step) + ))[0] From 893096c9bbb21f765636ed2f9d09c434fd0d2842 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 21 May 2026 11:53:36 +0200 Subject: [PATCH 04/11] updated compaction of ld_lists --- src/hermes/model/types/ld_container.py | 2 +- src/hermes/model/types/ld_list.py | 41 ++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/src/hermes/model/types/ld_container.py b/src/hermes/model/types/ld_container.py index abc37fa9..10d2a82c 100644 --- a/src/hermes/model/types/ld_container.py +++ b/src/hermes/model/types/ld_container.py @@ -316,7 +316,7 @@ def compact( COMPACTED_JSON_LD_VALUE: The compacted version of selfs JSON-LD representation. """ return self.ld_proc.compact( - self.ld_value, context or self.context, {"documentLoader": bundled_loader, "skipExpand": True} + self.ld_value, context or self.full_context, {"documentLoader": bundled_loader, "skipExpand": True} ) def to_python(self): diff --git a/src/hermes/model/types/ld_list.py b/src/hermes/model/types/ld_list.py index 003cda82..01a1c265 100644 --- a/src/hermes/model/types/ld_list.py +++ b/src/hermes/model/types/ld_list.py @@ -15,6 +15,7 @@ from typing_extensions import Self from .ld_container import ( + COMPACTED_JSON_LD_VALUE, ld_container, JSON_LD_CONTEXT_DICT, EXPANDED_JSON_LD_VALUE, @@ -23,6 +24,7 @@ TIME_TYPE, BASIC_TYPE, ) +from .pyld_util import bundled_loader if TYPE_CHECKING: from .ld_dict import ld_dict @@ -548,6 +550,45 @@ def to_python(self: Self) -> list[Union[BASIC_TYPE, TIME_TYPE, PYTHONIZED_LD_CON for item in self ] + def compact( + self: Self, context: Optional[Union[list[Union[JSON_LD_CONTEXT_DICT, str]], JSON_LD_CONTEXT_DICT, str]] = None + ) -> COMPACTED_JSON_LD_VALUE: + """ + Returns the compacted version of the given ld_list using its context only if none was supplied. + The returned object is of the form `{"@context": the_context, container_type: compacted_content}`. + + Args: + context (list[JSON_LD_CONTEXT_DICT | str] | JSON_LD_CONTEXT_DICT | str | None): + The context to use for the compaction. If None the context of self is used. + + Returns: + COMPACTED_JSON_LD_VALUE: The compacted version of selfs JSON-LD representation. + """ + # compact the ld_list standalone if necessary + if self.key is None: + return self.ld_proc.compact( + self.ld_value, context or self.full_context, {"documentLoader": bundled_loader, "skipExpand": True} + ) + # compact the ld_list within a temporary dictionary + temp_dict = self.ld_proc.compact( + [{self.ld_proc.expand_iri(self.active_ctx, self.key): self.ld_value}], + context or self.full_context, + {"documentLoader": bundled_loader, "skipExpand": True} + ) + context = temp_dict["@context"] + temp_container = temp_dict[ + self.ld_proc.compact_iri(self.active_ctx, self.ld_proc.expand_iri(self.active_ctx, self.key)) + ] + if self.container_type != "@set": + return { + "@context": context, + **temp_container + } + return { + "@context": context, + "@set": temp_container if isinstance(temp_container, list) else [temp_container] + } + @classmethod def is_ld_list(cls: type[Self], ld_value: Any) -> bool: """ From 4686353f5fda49ca3c5434b2e904b51f8690d6de Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 21 May 2026 13:25:18 +0200 Subject: [PATCH 05/11] fixed provenance document loading in harvest --- src/hermes/commands/harvest/base.py | 5 ++--- src/hermes/model/provenance/ld_prov.py | 19 +++++++++++++++++-- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index e6fbfc4c..30c00f23 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -139,7 +139,7 @@ def __call__(self, args: argparse.Namespace) -> None: }) with ctx["provenance"] as cache: - cache["codemeta"] = prov_doc.ld_value + cache["result"] = prov_doc.ld_value ctx.finalize_step('harvest') if not harvested_any: @@ -151,8 +151,7 @@ def init_provenance_document(self) -> ld_prov_list: 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 + return ld_prov_list.load_ld_prov_list(cache["result"]) except KeyError: pass prov_doc = ld_prov_list() diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index df4b8788..4ce93a4b 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -25,14 +25,29 @@ class ld_prov_list(ld_list): 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: - super().__init__([{"@graph": []}], parent=parent, key=key, index=index, context=context) + super().__init__(data, parent=parent, key=key, index=index, context=context) + + @classmethod + def load_ld_prov_list(cls, data) -> "ld_prov_list": + if cls.INDICES != {}: + raise RuntimeError("Only zero or one objects of class 'ld_prov_list' may exist at every point in time.") + prov_list = cls.from_list(data[0]["@graph"], container_type="@graph", context=ALL_CONTEXTS, key=cls.PROV_DOC_IRI) + for item in prov_list: + if not ("@id" in item and item["@id"].startswith("_:")): + continue + item_id = item["@id"][2:].split("/") + if not (len(item_id) == 2 and item_id[1].isnumeric()): + continue + if cls.INDICES.get(item_id[0], 0) < int(item_id[1]): + cls.INDICES[item_id[0]] = int(item_id[1]) + return prov_list def next_node_iri(self, type) -> str: if type not in ld_prov_list.INDICES: From c05601b7753a7cb6619983475c87ca602879d723 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 21 May 2026 13:29:33 +0200 Subject: [PATCH 06/11] flake8 --- src/hermes/commands/harvest/base.py | 1 - src/hermes/model/provenance/ld_prov.py | 4 +++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 30c00f23..f1ae63b7 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -13,7 +13,6 @@ from hermes.model.context_manager import HermesContext from hermes.model import SoftwareMetadata from hermes.model.provenance.ld_prov import ld_prov_list -from hermes.model.types.ld_context import ALL_CONTEXTS class HermesHarvestPlugin(HermesPlugin): diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index 4ce93a4b..3c6ea7c3 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -38,7 +38,9 @@ def __init__( def load_ld_prov_list(cls, data) -> "ld_prov_list": if cls.INDICES != {}: raise RuntimeError("Only zero or one objects of class 'ld_prov_list' may exist at every point in time.") - prov_list = cls.from_list(data[0]["@graph"], container_type="@graph", context=ALL_CONTEXTS, key=cls.PROV_DOC_IRI) + prov_list = cls.from_list( + data[0]["@graph"], key=cls.PROV_DOC_IRI, context=ALL_CONTEXTS, container_type="@graph" + ) for item in prov_list: if not ("@id" in item and item["@id"].startswith("_:")): continue From 379fe0fbd90731f0c288f3ab85ac08627c13d0b4 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 21 May 2026 16:31:45 +0200 Subject: [PATCH 07/11] first draft of provenance recording for process step --- src/hermes/commands/harvest/base.py | 18 ++-- src/hermes/commands/process/base.py | 112 ++++++++++++++++++++++++- src/hermes/model/provenance/ld_prov.py | 12 +-- 3 files changed, 128 insertions(+), 14 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index f1ae63b7..51744a50 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -128,13 +128,19 @@ def __call__(self, args: argparse.Namespace) -> None: }) # TODO: add more info prov_doc.add_entity(data={ - "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": data_output.ref, + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref }) prov_doc.add_entity(data={ - "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": data_output.ref, + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref }) prov_doc.add_entity(data={ - "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": data_output.ref, + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref }) with ctx["provenance"] as cache: @@ -162,7 +168,7 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N if plugin is None: return # two passes are needed because the nodes are nested exactly two levels - related = prov_doc.shallow_search(lambda doc, node: ( + related = prov_doc.shallow_search(lambda node: ( ("prov:wasAssociatedWith" in node and plugin.ref in node["prov:wasAssociatedWith"]) or ("prov:wasAttributedTo" in node and plugin.ref in node["prov:wasAttributedTo"]) )) @@ -170,11 +176,11 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N del prov_doc[plugin.index] return ids = [plugin.ref, *(rel.ref for rel in related)] - related = prov_doc.shallow_search(lambda doc, node: any( + related = prov_doc.shallow_search(lambda node: any( (f"prov:{key}" in node and 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(lambda doc, node: ("@id" in node and node["@id"] == item["@id"])) + items = prov_doc.shallow_search(lambda node: ("@id" in node and node["@id"] == item["@id"])) del prov_doc[items[0].index] diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py index 725f6487..a603c650 100644 --- a/src/hermes/commands/process/base.py +++ b/src/hermes/commands/process/base.py @@ -5,7 +5,7 @@ # SPDX-FileContributor: Michael Meinel import argparse -from typing import Union +from typing import Optional from pydantic import BaseModel @@ -15,12 +15,13 @@ from hermes.model.context_manager import HermesContext from hermes.model.merge.action import MergeAction from hermes.model.merge.container import ld_merge_dict +from hermes.model.provenance.ld_prov import ld_prov_list class HermesProcessPlugin(HermesPlugin): """ Base plugin that defines additional merge strategies.""" - def __call__(self, command: HermesCommand) -> dict[Union[str, None], dict[Union[str, None], MergeAction]]: + def __call__(self, command: HermesCommand) -> dict[Optional[str], dict[Optional[str], MergeAction]]: pass @@ -38,6 +39,12 @@ class HermesProcessCommand(HermesCommand): settings_class = ProcessSettings def __call__(self, args: argparse.Namespace) -> None: + self.log.info("# Load provenance data from harvest step") + prov_doc = self.load_prov_doc() + if prov_doc is not None: + process_command = prov_doc.get_hermes_command("process") + hermes_cache = prov_doc.get_hermes_cache() + self.log.info("# Metadata processing") merged_doc = ld_merge_dict([{}]) @@ -56,6 +63,7 @@ def __call__(self, args: argparse.Namespace) -> None: self.log.info("## Load and run the plugins") any_strategies_loaded = False + strategy_action, merged_strategies = None, None # add the strategies from the plugins for plugin_name in reversed(self.settings.plugins): self.log.info(f"### Load {plugin_name} plugin") @@ -79,6 +87,28 @@ def __call__(self, args: argparse.Namespace) -> None: merged_doc.add_strategy(additional_strategies) any_strategies_loaded = True + if prov_doc is None: + continue + plugin = prov_doc.add_hermes_plugin("process", plugin_name) + new_strategy_generation = prov_doc.add_activity(data={"prov:wasAssociatedWith": plugin.ref}) + new_strategies = prov_doc.add_entity( + data={"prov:wasAttributedTo": plugin.ref, "prov:wasGeneratedBy": new_strategy_generation.ref} + ) + if merged_strategies is None: + merged_strategies = new_strategies + strategy_action = new_strategy_generation + continue + strategy_action = prov_doc.add_activity(data={ + "prov:used": [merged_strategies.ref, new_strategies.ref], + "prov:wasInformedBy": [strategy_action.ref, new_strategy_generation.ref], + "prov:wasAssociatedWith": process_command.ref + }) + merged_strategies = prov_doc.add_entity(data={ + "prov:wasDerivedFrom": [merged_strategies.ref, new_strategies.ref], + "prov:wasGeneratedBy": strategy_action.ref, + "prov:wasAttributedTo": process_command.ref + }) + if not any_strategies_loaded: self.log.critical("## No process plugin was ran successfully.") raise HermesPluginRunError("No process plugin was ran successfully.") @@ -89,6 +119,7 @@ def __call__(self, args: argparse.Namespace) -> None: # merge data from harvesters self.log.info("## Merge the metadata of the harvesters") merged_any = False + merge_action, merged_data = None, None for harvester in harvester_names: self.log.info(f"### Load data from {harvester} plugin") # load data from harvester @@ -110,6 +141,43 @@ def __call__(self, args: argparse.Namespace) -> None: raise RuntimeError(f"Merging the data from {harvester} plugin failed.") from e merged_any = True + if prov_doc is None: + continue + harvest_plugin = prov_doc.get_hermes_plugin("harvest", harvester) + harvest_command = prov_doc.get_hermes_command("harvest") + store_action = prov_doc.shallow_search(lambda node: ( + "prov:wasAssociatedWith" in node and + node["prov:wasAssociatedWith"] == [harvest_plugin.ref, hermes_cache.ref, harvest_command.ref] + ))[0] + stored_results = [ + result.ref for result in prov_doc.shallow_search( + lambda node: ("prov:wasGeneratedBy" in node and node["prov:wasGeneratedBy"] == store_action.ref) + ) + ] + new_data_load = prov_doc.add_activity(data={ + "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref], + "prov:used": stored_results + }) + new_data = prov_doc.add_entity(data={ + "prov:wasAttributedTo": plugin.ref, + "prov:wasGeneratedBy": new_data_load.ref, + "prov:wasDerivedFrom": stored_results + }) + if merged_data is None: + merged_data = new_data + merge_action = new_data_load + continue + merge_action = prov_doc.add_activity(data={ + "prov:used": [merged_data.ref, new_data.ref, merged_strategies.ref], + "prov:wasInformedBy": [merge_action.ref, new_data_load.ref], + "prov:wasAssociatedWith": process_command.ref + }) + merged_data = prov_doc.add_entity(data={ + "prov:wasDerivedFrom": [merged_data.ref, new_data.ref], + "prov:wasGeneratedBy": merge_action.ref, + "prov:wasAttributedTo": process_command.ref + }) + # error if nothing was merged if not merged_any: self.log.critical("No metadata has been merged, the loading of the data failed for all harvesters.") @@ -122,6 +190,46 @@ def __call__(self, args: argparse.Namespace) -> None: result_ctx["codemeta"] = merged_doc.compact() result_ctx["context"] = {"@context": merged_doc.full_context} result_ctx["expanded"] = merged_doc.ld_value + + if prov_doc is not None: + write = prov_doc.add_activity(data={ + "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref, plugin.ref], + "prov:used": merged_data.ref, + "prov:wasInformedBy": merge_action.ref + }) + # TODO: add more info + prov_doc.add_entity(data={ + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": merged_data.ref, + "prov:wasAttributedTo": hermes_cache.ref + }) + prov_doc.add_entity(data={ + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": merged_data.ref, + "prov:wasAttributedTo": hermes_cache.ref + }) + prov_doc.add_entity(data={ + "prov:wasGeneratedBy": write.ref, + "prov:wasDerivedFrom": merged_data.ref, + "prov:wasAttributedTo": hermes_cache.ref + }) + + with ctx["provenance"] as cache: + cache["result"] = prov_doc.ld_value + ctx.finalize_step("process") ctx.finalize_step("harvest") + + def load_prov_doc(self) -> Optional[ld_prov_list]: + ctx = HermesContext() + ctx.prepare_step("harvest") + with ctx["provenance"] as cache: + try: + return ld_prov_list.load_ld_prov_list(cache["result"]) + except Exception: + self.log.warning( + "The provenance data from the harvest step could not be loaded." + "Processing will proceed without collecting provenance data.", + exc_info=1 + ) diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index 3c6ea7c3..9edc4019 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -132,21 +132,21 @@ def add_hermes_plugin(self, step, name) -> ld_dict: return node def shallow_search(self, query) -> list[ld_dict]: - return [item for item in self if query(self, item)] + return [item for item in self if query(item)] def get_hermes(self) -> ld_dict: - return self.shallow_search(lambda doc, node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_ID))[0] + return self.shallow_search(lambda node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_ID))[0] def get_hermes_cache(self) -> ld_dict: - return self.shallow_search(lambda doc, node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_CACHE_ID))[0] + return self.shallow_search(lambda node: ("@id" in node and node["@id"] == ld_prov_list.HERMES_CACHE_ID))[0] def get_hermes_base_plugin(self, step) -> ld_dict: - return self.shallow_search(lambda doc, node: ( + return self.shallow_search(lambda node: ( "@id" in node and node["@id"] == ld_prov_list.HERMES_BASE_PLUGIN_ID_FORMAT.format(step=step) ))[0] def get_hermes_plugin(self, step, name) -> Union[ld_dict, None]: - search_result = self.shallow_search(lambda doc, node: ( + search_result = self.shallow_search(lambda node: ( "@id" in node and node["@id"] == ld_prov_list.HERMES_PLUGIN_ID_FORMAT.format(step=step, name=name) )) if search_result: @@ -154,6 +154,6 @@ def get_hermes_plugin(self, step, name) -> Union[ld_dict, None]: return None def get_hermes_command(self, step) -> ld_dict: - return self.shallow_search(lambda doc, node: ( + return self.shallow_search(lambda node: ( "@id" in node and node["@id"] == ld_prov_list.HERMES_COMMAND_ID_FORMAT.format(step=step) ))[0] From 40a848ab1abccd0ca1e6a5c1fce0d8c6b7a3a68e Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 28 May 2026 14:13:28 +0200 Subject: [PATCH 08/11] improved provenance collection for harvest --- src/hermes/commands/harvest/base.py | 47 +++++++++++++++++++++----- src/hermes/commands/process/base.py | 2 +- src/hermes/model/provenance/ld_prov.py | 12 +++---- 3 files changed, 46 insertions(+), 15 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 51744a50..f720f8b7 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -5,6 +5,7 @@ # SPDX-FileContributor: Michael Meinel import argparse +import datetime from pydantic import BaseModel @@ -28,9 +29,11 @@ def __call__(self, command: HermesCommand) -> SoftwareMetadata: pass def load(): + # TODO: Implement pass def write(): + # TODO: Implement pass @@ -75,14 +78,17 @@ def __call__(self, args: argparse.Namespace) -> None: self.log.info(f"### Run {plugin_name} plugin") # run plugin try: - harvested_data = plugin_func(self) + harvested_data: SoftwareMetadata = plugin_func(self) except Exception: self.log.exception(f"### Unknown error while executing the {plugin_name} plugin, skipping it now.") continue + returned_at_time = datetime.datetime.now().isoformat() self.log.info(f"### Store metadata harvested by {plugin_name} plugin") # store harvested data + begin_store_at_time = datetime.datetime.now().isoformat() harvested_data.write_to_cache(ctx, plugin_name) + stored_at_time = datetime.datetime.now().isoformat() harvested_any = True self.remove_provenance_info_for_plugin(prov_doc, plugin_name) @@ -107,40 +113,64 @@ def __call__(self, args: argparse.Namespace) -> None: io_ops.append(io_op.ref) map_activity = prov_doc.add_activity(data={ + "schema:description": "Maps the loaded data to the JSON-LD contexts vocabulary.", "prov:wasInformedBy": io_ops, "prov:used": outputs, - "prov:wasAssociatedWith": plugin.ref + "prov:wasAssociatedWith": plugin.ref, + "prov:startedAtTime": returned_at_time }) data_output = prov_doc.add_entity(data={ + "schema:description": "the harvested metadata", "prov:wasAttributedTo": plugin.ref, "prov:wasGeneratedBy": map_activity.ref, - "prov:wasDerivedFrom": outputs + "prov:wasDerivedFrom": outputs, + "prov:generatedAtTime": returned_at_time }) write = prov_doc.add_activity(data={ + "schema:description": "Writes the harvested metadata into the HERMES cache.", "prov:wasAssociatedWith": [ prov_doc.get_hermes_command("harvest").ref, prov_doc.get_hermes_cache().ref, plugin.ref ], "prov:used": data_output.ref, - "prov:wasInformedBy": map_activity.ref + "prov:wasInformedBy": map_activity.ref, + "prov:startedAtTime": begin_store_at_time, + "prov:endedAtTime": stored_at_time }) - # TODO: add more info prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The compacted version of the harvested metadata.", + "schema:text": str(harvested_data.compact()), + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "codemeta.json").as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, - "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, + "prov:generatedAtTime": stored_at_time }) prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The context of the harvested metadata.", + "schema:text": str({"@context": harvested_data.full_context}), + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "context.json").as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, - "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, + "prov:generatedAtTime": stored_at_time }) prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The expanded version of the harvested metadata.", + "schema:text": str(harvested_data.ld_value), + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "expanded.json").as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, - "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, + "prov:generatedAtTime": stored_at_time }) with ctx["provenance"] as cache: @@ -181,6 +211,7 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N "wasAssociatedWith", "wasAttributedTo", "wasGeneratedBy", "used", "wasDerivedFrom", "wasInformedBy" ] )) + del prov_doc[plugin.index] for item in related: items = prov_doc.shallow_search(lambda node: ("@id" in node and node["@id"] == item["@id"])) del prov_doc[items[0].index] diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py index a603c650..51bac589 100644 --- a/src/hermes/commands/process/base.py +++ b/src/hermes/commands/process/base.py @@ -229,7 +229,7 @@ def load_prov_doc(self) -> Optional[ld_prov_list]: return ld_prov_list.load_ld_prov_list(cache["result"]) except Exception: self.log.warning( - "The provenance data from the harvest step could not be loaded." + "The provenance data from the harvest step could not be loaded. " "Processing will proceed without collecting provenance data.", exc_info=1 ) diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index 9edc4019..6604df1c 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -61,9 +61,9 @@ def add_activity(self, *, data={}) -> ld_dict: self.append(data) activity = self[-1] if "@type" not in data: - activity["@type"] = "prov:Activity" + activity["@type"] = ["prov:Activity", "schema:Action"] else: - activity["@type"].append("prov:Activity") + activity["@type"].extend(["prov:Activity", "schema:Action"]) if "@id" not in data: activity["@id"] = self.next_node_iri("Activity") return activity @@ -72,9 +72,9 @@ def add_agent(self, *, data={}) -> ld_dict: self.append(data) agent = self[-1] if "@type" not in data: - agent["@type"] = "prov:Agent" + agent["@type"] = ["prov:Agent", "schema:SoftwareApplication"] else: - agent["@type"].append("prov:Agent") + agent["@type"].extend(["prov:Agent", "schema:SoftwareApplication"]) if "@id" not in data: agent["@id"] = self.next_node_iri("Agent") return agent @@ -83,9 +83,9 @@ def add_entity(self, *, data={}) -> ld_dict: self.append(data) entity = self[-1] if "@type" not in data: - entity["@type"] = "prov:Entity" + entity["@type"] = ["prov:Entity", "schema:Thing"] else: - entity["@type"].append("prov:Entity") + entity["@type"].extend(["prov:Entity", "schema:Thing"]) if "@id" not in data: entity["@id"] = self.next_node_iri("Entity") return entity From d637f84832f6bf2944ce2a731ca8a1d34ba75c2b Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Mon, 1 Jun 2026 16:26:14 +0200 Subject: [PATCH 09/11] implement load wrapper for harvest plugins and fixed minor bug --- src/hermes/commands/harvest/base.py | 39 ++++++++++++++++++------- src/hermes/commands/harvest/cff.py | 2 +- src/hermes/commands/harvest/codemeta.py | 2 +- 3 files changed, 31 insertions(+), 12 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index f720f8b7..7545c52f 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -6,6 +6,8 @@ import argparse import datetime +from io import IOBase +from pathlib import Path from pydantic import BaseModel @@ -28,9 +30,24 @@ def __init__(self): def __call__(self, command: HermesCommand) -> SoftwareMetadata: pass - def load(): - # TODO: Implement - pass + def load(self, func, source, *args, **kwargs): + source_metadata = {"schema:description": "metadata source"} + if isinstance(source, IOBase): + source_metadata["schema:url"] = Path(source.name).absolute().as_uri() + elif isinstance(source, Path): + source_metadata["schema:url"] = source.absolute().as_uri() + elif isinstance(source, str): + source_metadata["schema:url"] = Path(source).absolute().as_uri() + io_operation = { + "schema:description": "Load operation called with (" + f"{source_metadata['schema:url'] if 'schema:url' in source_metadata else str(source)}" + f"{', ' + str(args) if args else ''}{', ' + str(kwargs) if kwargs else ''}).", + "schema:name": f"{func.__module__}.{func.__qualname__}" + } + result = func(source, *args, **kwargs) + loaded_metadata = {"schema:description": "the loaded data", "schema:text": str(result)} + self.io_operations.append((source_metadata, io_operation, loaded_metadata)) + return result def write(): # TODO: Implement @@ -108,7 +125,7 @@ def __call__(self, args: argparse.Namespace) -> None: "prov:wasDerivedFrom": loaded_source.ref, "prov:wasGeneratedBy": io_op.ref }) - loaded_data = prov_doc.add_entity(data=plugin_io_operations[2]) + loaded_data = prov_doc.add_entity(data=plugin_io_operation[2]) outputs.append(loaded_data.ref) io_ops.append(io_op.ref) @@ -144,7 +161,7 @@ def __call__(self, args: argparse.Namespace) -> None: "schema:description": "The compacted version of the harvested metadata.", "schema:text": str(harvested_data.compact()), "schema:encodingFormat": "application/json", - "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "codemeta.json").as_uri(), + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "codemeta.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, @@ -155,7 +172,7 @@ def __call__(self, args: argparse.Namespace) -> None: "schema:description": "The context of the harvested metadata.", "schema:text": str({"@context": harvested_data.full_context}), "schema:encodingFormat": "application/json", - "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "context.json").as_uri(), + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "context.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, @@ -166,7 +183,7 @@ def __call__(self, args: argparse.Namespace) -> None: "schema:description": "The expanded version of the harvested metadata.", "schema:text": str(harvested_data.ld_value), "schema:encodingFormat": "application/json", - "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "expanded.json").as_uri(), + "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "expanded.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": data_output.ref, "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, @@ -197,7 +214,6 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N plugin = prov_doc.get_hermes_plugin("harvest", plugin) if plugin is None: return - # two passes are needed because the nodes are nested exactly two levels related = prov_doc.shallow_search(lambda node: ( ("prov:wasAssociatedWith" in node and plugin.ref in node["prov:wasAssociatedWith"]) or ("prov:wasAttributedTo" in node and plugin.ref in node["prov:wasAttributedTo"]) @@ -206,7 +222,9 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N del prov_doc[plugin.index] return ids = [plugin.ref, *(rel.ref for rel in related)] - related = prov_doc.shallow_search(lambda node: any( + used_entities = [rel["prov:used"][0]["@id"] for rel in related if "prov:used" in rel] + related = prov_doc.shallow_search(lambda node: node["@id"] in used_entities) + related += prov_doc.shallow_search(lambda node: any( (f"prov:{key}" in node and id in node[f"prov:{key}"]) for id in ids for key in [ "wasAssociatedWith", "wasAttributedTo", "wasGeneratedBy", "used", "wasDerivedFrom", "wasInformedBy" ] @@ -214,4 +232,5 @@ def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> N del prov_doc[plugin.index] for item in related: items = prov_doc.shallow_search(lambda node: ("@id" in node and node["@id"] == item["@id"])) - del prov_doc[items[0].index] + if len(items) == 1: + del prov_doc[items[0].index] diff --git a/src/hermes/commands/harvest/cff.py b/src/hermes/commands/harvest/cff.py index 5a2d16c1..f2b648e6 100644 --- a/src/hermes/commands/harvest/cff.py +++ b/src/hermes/commands/harvest/cff.py @@ -43,7 +43,7 @@ def __call__(self, command: HermesHarvestCommand) -> tuple[SoftwareMetadata, dic 'Aborting harvesting for this metadata source.') # Read the content - cff_data = cff_file.read_text() + cff_data = self.load(pathlib.Path.read_text, cff_file) cff_dict = self._load_cff_from_file(cff_data) if command.settings.cff.enable_validation: diff --git a/src/hermes/commands/harvest/codemeta.py b/src/hermes/commands/harvest/codemeta.py index 3dc84296..07645647 100644 --- a/src/hermes/commands/harvest/codemeta.py +++ b/src/hermes/commands/harvest/codemeta.py @@ -34,7 +34,7 @@ def __call__(self, command: HermesHarvestCommand) -> tuple[SoftwareMetadata, dic ) # Read the content - codemeta_str = codemeta_file.read_text() + codemeta_str = self.load(pathlib.Path.read_text, codemeta_file) if not self._validate(codemeta_file): raise HermesValidationError(codemeta_file) From 4d8e62da8d8ba024e7dc024318c972169ec2e9fc Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Mon, 8 Jun 2026 17:34:38 +0200 Subject: [PATCH 10/11] implemented draft of provenance collection druing merge --- src/hermes/commands/harvest/base.py | 65 +++++++++---------- src/hermes/commands/process/base.py | 88 +++++++++++++------------- src/hermes/model/merge/container.py | 51 ++++++++++++++- src/hermes/model/provenance/ld_prov.py | 2 +- 4 files changed, 127 insertions(+), 79 deletions(-) diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 7545c52f..56799e9c 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -38,10 +38,10 @@ def load(self, func, source, *args, **kwargs): source_metadata["schema:url"] = source.absolute().as_uri() elif isinstance(source, str): source_metadata["schema:url"] = Path(source).absolute().as_uri() - io_operation = { + io_operation = { "schema:description": "Load operation called with (" - f"{source_metadata['schema:url'] if 'schema:url' in source_metadata else str(source)}" - f"{', ' + str(args) if args else ''}{', ' + str(kwargs) if kwargs else ''}).", + f"{source_metadata['schema:url'] if 'schema:url' in source_metadata else str(source)}" + f"{', ' + str(args) if args else ''}{', ' + str(kwargs) if kwargs else ''}).", "schema:name": f"{func.__module__}.{func.__qualname__}" } result = func(source, *args, **kwargs) @@ -60,6 +60,32 @@ class HarvestSettings(BaseModel): sources: list[str] = [] +def remove_harvest_plugin_from_prov_doc(prov_doc: ld_prov_list, plugin: str) -> None: + plugin = prov_doc.get_hermes_plugin("harvest", plugin) + if plugin is None: + return + related = prov_doc.shallow_search(lambda node: ( + ("prov:wasAssociatedWith" in node and plugin.ref in node["prov:wasAssociatedWith"]) or + ("prov:wasAttributedTo" in node and plugin.ref in node["prov:wasAttributedTo"]) + )) + if len(related) == 0: + del prov_doc[plugin.index] + return + ids = [plugin.ref, *(rel.ref for rel in related)] + used_entities = [rel["prov:used"][0]["@id"] for rel in related if "prov:used" in rel] + related = prov_doc.shallow_search(lambda node: node["@id"] in used_entities) + related += prov_doc.shallow_search(lambda node: any( + (f"prov:{key}" in node and id in node[f"prov:{key}"]) for id in ids for key in [ + "wasAssociatedWith", "wasAttributedTo", "wasGeneratedBy", "used", "wasDerivedFrom", "wasInformedBy" + ] + )) + del prov_doc[plugin.index] + for item in related: + items = prov_doc.shallow_search(lambda node: ("@id" in node and node["@id"] == item["@id"])) + if len(items) == 1: + del prov_doc[items[0].index] + + class HermesHarvestCommand(HermesCommand): """ Harvest metadata from configured sources. """ @@ -108,7 +134,7 @@ def __call__(self, args: argparse.Namespace) -> None: stored_at_time = datetime.datetime.now().isoformat() harvested_any = True - self.remove_provenance_info_for_plugin(prov_doc, plugin_name) + remove_harvest_plugin_from_prov_doc(prov_doc, plugin_name) plugin = prov_doc.add_hermes_plugin("harvest", plugin_name) plugin_io_operations = plugin_func.io_operations @@ -159,7 +185,7 @@ def __call__(self, args: argparse.Namespace) -> None: prov_doc.add_entity(data={ "@type": "schema:CreativeWork", "schema:description": "The compacted version of the harvested metadata.", - "schema:text": str(harvested_data.compact()), + "schema:text": str(harvested_data.compact()), # TODO: maybe "prov:value" instead? "schema:encodingFormat": "application/json", "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "codemeta.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, @@ -170,7 +196,7 @@ def __call__(self, args: argparse.Namespace) -> None: prov_doc.add_entity(data={ "@type": "schema:CreativeWork", "schema:description": "The context of the harvested metadata.", - "schema:text": str({"@context": harvested_data.full_context}), + "schema:text": str({"@context": harvested_data.full_context}), # TODO: maybe "prov:value" instead? "schema:encodingFormat": "application/json", "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "context.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, @@ -181,7 +207,7 @@ def __call__(self, args: argparse.Namespace) -> None: prov_doc.add_entity(data={ "@type": "schema:CreativeWork", "schema:description": "The expanded version of the harvested metadata.", - "schema:text": str(harvested_data.ld_value), + "schema:text": str(harvested_data.ld_value), # TODO: maybe "prov:value" instead? "schema:encodingFormat": "application/json", "schema:url": (ctx.cache_dir / "harvest" / plugin_name / "expanded.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, @@ -209,28 +235,3 @@ def init_provenance_document(self) -> ld_prov_list: prov_doc = ld_prov_list() prov_doc.init_hermes_agents() return prov_doc - - def remove_provenance_info_for_plugin(self, prov_doc: ld_prov_list, plugin) -> None: - plugin = prov_doc.get_hermes_plugin("harvest", plugin) - if plugin is None: - return - related = prov_doc.shallow_search(lambda node: ( - ("prov:wasAssociatedWith" in node and plugin.ref in node["prov:wasAssociatedWith"]) or - ("prov:wasAttributedTo" in node and plugin.ref in node["prov:wasAttributedTo"]) - )) - if len(related) == 0: - del prov_doc[plugin.index] - return - ids = [plugin.ref, *(rel.ref for rel in related)] - used_entities = [rel["prov:used"][0]["@id"] for rel in related if "prov:used" in rel] - related = prov_doc.shallow_search(lambda node: node["@id"] in used_entities) - related += prov_doc.shallow_search(lambda node: any( - (f"prov:{key}" in node and id in node[f"prov:{key}"]) for id in ids for key in [ - "wasAssociatedWith", "wasAttributedTo", "wasGeneratedBy", "used", "wasDerivedFrom", "wasInformedBy" - ] - )) - del prov_doc[plugin.index] - for item in related: - items = prov_doc.shallow_search(lambda node: ("@id" in node and node["@id"] == item["@id"])) - if len(items) == 1: - del prov_doc[items[0].index] diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py index 51bac589..5711b197 100644 --- a/src/hermes/commands/process/base.py +++ b/src/hermes/commands/process/base.py @@ -10,6 +10,7 @@ from pydantic import BaseModel from hermes.commands.base import HermesCommand, HermesPlugin +from hermes.commands.harvest.base import remove_harvest_plugin_from_prov_doc from hermes.error import HermesPluginRunError, MisconfigurationError from hermes.model.api import SoftwareMetadata from hermes.model.context_manager import HermesContext @@ -46,7 +47,7 @@ def __call__(self, args: argparse.Namespace) -> None: hermes_cache = prov_doc.get_hermes_cache() self.log.info("# Metadata processing") - merged_doc = ld_merge_dict([{}]) + merged_doc = ld_merge_dict([{}], prov_doc) if not self.settings.plugins: self.log.critical( @@ -119,7 +120,6 @@ def __call__(self, args: argparse.Namespace) -> None: # merge data from harvesters self.log.info("## Merge the metadata of the harvesters") merged_any = False - merge_action, merged_data = None, None for harvester in harvester_names: self.log.info(f"### Load data from {harvester} plugin") # load data from harvester @@ -127,56 +127,56 @@ def __call__(self, args: argparse.Namespace) -> None: metadata = SoftwareMetadata.load_from_cache(ctx, harvester) except Exception: # skip this harvester when the data is invalid + if prov_doc is not None: + remove_harvest_plugin_from_prov_doc(prov_doc, harvester) self.log.exception( f"### The data from the harvester {harvester} could not be loaded or is invalid, skipping it now." ) continue + if prov_doc is not None: + harvest_plugin = prov_doc.get_hermes_plugin("harvest", harvester) + harvest_command = prov_doc.get_hermes_command("harvest") + store_action = prov_doc.shallow_search(lambda node: ( + "prov:wasAssociatedWith" in node and + node["prov:wasAssociatedWith"] == [harvest_plugin.ref, hermes_cache.ref, harvest_command.ref] + ))[0] + stored_results = [ + result.ref for result in prov_doc.shallow_search(lambda node: ( + "prov:wasGeneratedBy" in node and node["prov:wasGeneratedBy"] == [store_action.ref] + )) + ] + new_action = prov_doc.add_activity(data={ # load of new data + "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref], + "prov:used": stored_results + }) + new_data = prov_doc.add_entity(data={ # new data to be merged + "prov:wasAttributedTo": plugin.ref, + "prov:wasGeneratedBy": new_action.ref, + "prov:wasDerivedFrom": stored_results + }) + if merged_any: + # One pass must have been completed already. + new_action = prov_doc.add_activity(data={ + "prov:used": [last_data.ref, new_data.ref], + "prov:wasInformedBy": [last_action.ref, new_action.ref], + "prov:wasAssociatedWith": process_command.ref + }) # initial merge action of the merge + merged_doc.prov_objects = [new_action, new_data, last_data] + self.log.info(f"### Merge data from {harvester} plugin") # merge data into the merge dict try: merged_doc.update(metadata) except Exception as e: + # TODO: Maybe this state is recoverable by starting over again and skipping this plugin. self.log.critical(f"### Merging the data from {harvester} plugin resulted in an error.", exc_info=True) raise RuntimeError(f"Merging the data from {harvester} plugin failed.") from e - merged_any = True - if prov_doc is None: - continue - harvest_plugin = prov_doc.get_hermes_plugin("harvest", harvester) - harvest_command = prov_doc.get_hermes_command("harvest") - store_action = prov_doc.shallow_search(lambda node: ( - "prov:wasAssociatedWith" in node and - node["prov:wasAssociatedWith"] == [harvest_plugin.ref, hermes_cache.ref, harvest_command.ref] - ))[0] - stored_results = [ - result.ref for result in prov_doc.shallow_search( - lambda node: ("prov:wasGeneratedBy" in node and node["prov:wasGeneratedBy"] == store_action.ref) - ) - ] - new_data_load = prov_doc.add_activity(data={ - "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref], - "prov:used": stored_results - }) - new_data = prov_doc.add_entity(data={ - "prov:wasAttributedTo": plugin.ref, - "prov:wasGeneratedBy": new_data_load.ref, - "prov:wasDerivedFrom": stored_results - }) - if merged_data is None: - merged_data = new_data - merge_action = new_data_load - continue - merge_action = prov_doc.add_activity(data={ - "prov:used": [merged_data.ref, new_data.ref, merged_strategies.ref], - "prov:wasInformedBy": [merge_action.ref, new_data_load.ref], - "prov:wasAssociatedWith": process_command.ref - }) - merged_data = prov_doc.add_entity(data={ - "prov:wasDerivedFrom": [merged_data.ref, new_data.ref], - "prov:wasGeneratedBy": merge_action.ref, - "prov:wasAttributedTo": process_command.ref - }) + if prov_doc is not None: + last_action = merged_doc.prov_objects[0] if merged_any else new_action + last_data = merged_doc.prov_objects[2] if merged_any else new_data + merged_any = True # error if nothing was merged if not merged_any: @@ -194,23 +194,23 @@ def __call__(self, args: argparse.Namespace) -> None: if prov_doc is not None: write = prov_doc.add_activity(data={ "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref, plugin.ref], - "prov:used": merged_data.ref, - "prov:wasInformedBy": merge_action.ref + "prov:used": last_data.ref, + "prov:wasInformedBy": last_action.ref }) # TODO: add more info prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, - "prov:wasDerivedFrom": merged_data.ref, + "prov:wasDerivedFrom": last_data.ref, "prov:wasAttributedTo": hermes_cache.ref }) prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, - "prov:wasDerivedFrom": merged_data.ref, + "prov:wasDerivedFrom": last_data.ref, "prov:wasAttributedTo": hermes_cache.ref }) prov_doc.add_entity(data={ "prov:wasGeneratedBy": write.ref, - "prov:wasDerivedFrom": merged_data.ref, + "prov:wasDerivedFrom": last_data.ref, "prov:wasAttributedTo": hermes_cache.ref }) diff --git a/src/hermes/model/merge/container.py b/src/hermes/model/merge/container.py index e9cc03a2..a286c7e8 100644 --- a/src/hermes/model/merge/container.py +++ b/src/hermes/model/merge/container.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Callable, Optional, Union from typing_extensions import Self +from hermes.model.provenance.ld_prov import ld_prov_list from hermes.model.types import ld_container, ld_context, ld_dict, ld_list from hermes.model.types.ld_container import ( BASIC_TYPE, EXPANDED_JSON_LD_VALUE, JSON_LD_CONTEXT_DICT, JSON_LD_VALUE, TIME_TYPE @@ -50,6 +51,8 @@ def _to_python( if isinstance(value, ld_dict) and not isinstance(value, ld_merge_dict): value = ld_merge_dict( value.ld_value, + self.prov_doc, + self.prov_objects, parent=value.parent, key=value.key, index=value.index, @@ -60,6 +63,8 @@ def _to_python( if isinstance(value, ld_list) and not isinstance(value, ld_merge_list): value = ld_merge_list( value.ld_value, + self.prov_doc, + self.prov_objects, parent=value.parent, key=value.key, index=value.index, @@ -82,6 +87,8 @@ class ld_merge_list(_ld_merge_container, ld_list): def __init__( self: "ld_merge_list", data: Union[list[str], list[dict[str, EXPANDED_JSON_LD_VALUE]]], + prov_doc: ld_prov_list = None, + prov_objects: list[ld_dict] = 3*[None], *, parent: Optional[ld_container] = None, key: Optional[str] = None, @@ -108,6 +115,8 @@ def __init__( super().__init__(data, parent=parent, key=key, index=index, context=context) self.strategies = strategies + self.prov_doc = prov_doc + self.prov_objects = prov_objects class ld_merge_dict(_ld_merge_container, ld_dict): @@ -123,6 +132,8 @@ class ld_merge_dict(_ld_merge_container, ld_dict): def __init__( self: Self, data: list[dict[str, EXPANDED_JSON_LD_VALUE]], + prov_doc: ld_prov_list = None, + prov_objects: list[ld_dict] = 3*[None], *, parent: Optional[Union[ld_dict, ld_list]] = None, key: Optional[str] = None, @@ -154,6 +165,8 @@ def __init__( # add strategies self.strategies = strategies + self.prov_doc = prov_doc + self.prov_objects = prov_objects def update_context( self: Self, other_context: Union[list[Union[str, JSON_LD_CONTEXT_DICT]], None] @@ -225,9 +238,32 @@ def __setitem__(self: Self, key: str, value: Union[JSON_LD_VALUE, BASIC_TYPE, TI """ # create the new item if self[key] and value have to be merged. if key in self: - value = self._merge_item(key, value) + if self.prov_objects[0] is not None: + last_merged_data = self.prov_objects[2] + merge_activity, value = self._merge_item(key, value) + if self.prov_objects[0] is not None: + create_new_merged_data = last_merged_data is self.prov_objects[2] + elif self.prov_objects[0] is not None: + merge_activity = self.prov_doc.add_activity(data={ + "schema:name": "merge", + "schema:description": "foo", + "prov:used": {"@list": [self.prov_objects[2].ref, str(self.path+[key])]}, + "prov:wasInformedBy": self.prov_objects[0].ref + }) + create_new_merged_data = True # update the entry of self[key] super().__setitem__(key, value) + if self.prov_objects[0] is None: + return + self.prov_objects[0] = merge_activity + if create_new_merged_data: + self.prov_objects[2] = self.prov_doc.add_entity(data={ + "prov:wasAttributedTo": self.prov_doc.get_hermes_command("process").ref, + "prov:wasGeneratedBy": merge_activity.ref, + "prov:wasDerivedFrom": {"@list": [self.prov_objects[1].ref, self.prov_objects[2].ref]} + }) + else: + self.prov_objects[2]["prov:wasGeneratedBy"].append(merge_activity.ref) def match( self: Self, @@ -281,7 +317,18 @@ def _merge_item( merger = strategy.get(key, strategy.get(None, None)) if merger is None: raise MergeError(f"Can't merge, no strategy found for key '{key}'.") - return merger.merge(self, [*self.path, key], self[key], value) + if self.prov_objects[0] is not None: + merge_activity = self.prov_doc.add_activity(data={ + "schema:name": "merge", + "schema:description": "foo", + "prov:wasAssociatedWith": self.prov_doc.get_hermes_command("process").ref, + "prov:used": {"@list": [self.prov_objects[1].ref, self.prov_objects[2].ref, str(self.path+[key])]}, + "prov:wasInformedBy": self.prov_objects[0].ref + }) + self.prov_objects[0] = merge_activity + else: + merge_activity = None + return merge_activity, merger.merge(self, [*self.path, key], self[key], value) def _add_related( self: Self, rel: str, key: str, value: Union[BASIC_TYPE, TIME_TYPE, ld_dict, ld_list] diff --git a/src/hermes/model/provenance/ld_prov.py b/src/hermes/model/provenance/ld_prov.py index 6604df1c..767d58ad 100644 --- a/src/hermes/model/provenance/ld_prov.py +++ b/src/hermes/model/provenance/ld_prov.py @@ -127,7 +127,7 @@ def add_hermes_plugin(self, step, name) -> ld_dict: "@id": ld_prov_list.HERMES_PLUGIN_ID_FORMAT.format(step=step, name=name), "@type": "schema:SoftwareApplication", "schema:name": f"{utils.hermes_name} {step} plugin '{name}'", - "prov:actedOnBehalfOf": self.get_hermes_base_plugin(step) + "prov:actedOnBehalfOf": self.get_hermes_base_plugin(step).ref }) return node From 038b72a50c884944faeaa9019b094302936df301 Mon Sep 17 00:00:00 2001 From: notactuallyfinn Date: Thu, 2 Jul 2026 12:58:45 +0200 Subject: [PATCH 11/11] completed provenance recording in the process step at least for now --- src/hermes/commands/process/base.py | 56 +++++++++++++++++++++++------ src/hermes/model/merge/container.py | 24 +++++++++---- 2 files changed, 64 insertions(+), 16 deletions(-) diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py index 5711b197..b7d83a79 100644 --- a/src/hermes/commands/process/base.py +++ b/src/hermes/commands/process/base.py @@ -5,6 +5,7 @@ # SPDX-FileContributor: Michael Meinel import argparse +import datetime from typing import Optional from pydantic import BaseModel @@ -91,20 +92,27 @@ def __call__(self, args: argparse.Namespace) -> None: if prov_doc is None: continue plugin = prov_doc.add_hermes_plugin("process", plugin_name) - new_strategy_generation = prov_doc.add_activity(data={"prov:wasAssociatedWith": plugin.ref}) - new_strategies = prov_doc.add_entity( - data={"prov:wasAttributedTo": plugin.ref, "prov:wasGeneratedBy": new_strategy_generation.ref} - ) + new_strategy_generation = prov_doc.add_activity(data={ + "schema:description": "generate new merge strategies", + "prov:wasAssociatedWith": plugin.ref + }) + new_strategies = prov_doc.add_entity(data={ # TODO: record strategies + "schema:description": f"new merge strategies of plugin {plugin_name}", + "prov:wasAttributedTo": plugin.ref, + "prov:wasGeneratedBy": new_strategy_generation.ref + }) if merged_strategies is None: merged_strategies = new_strategies strategy_action = new_strategy_generation continue strategy_action = prov_doc.add_activity(data={ + "schema:description": "merging the new strategies into the others", "prov:used": [merged_strategies.ref, new_strategies.ref], "prov:wasInformedBy": [strategy_action.ref, new_strategy_generation.ref], "prov:wasAssociatedWith": process_command.ref }) - merged_strategies = prov_doc.add_entity(data={ + merged_strategies = prov_doc.add_entity(data={ # TODO: record strategies + "schema:description": "the merge strategies of multiple plugins merged together", "prov:wasDerivedFrom": [merged_strategies.ref, new_strategies.ref], "prov:wasGeneratedBy": strategy_action.ref, "prov:wasAttributedTo": process_command.ref @@ -147,10 +155,14 @@ def __call__(self, args: argparse.Namespace) -> None: )) ] new_action = prov_doc.add_activity(data={ # load of new data + "schema:description": f"loads the data from {harvester} plugin", "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref], "prov:used": stored_results }) new_data = prov_doc.add_entity(data={ # new data to be merged + "@type": "schema:CreativeWork", + "schema:description": f"data loaded from {harvester} plugin", + "schema:text": str(metadata.compact()), # TODO: maybe "prov:value" instead? "prov:wasAttributedTo": plugin.ref, "prov:wasGeneratedBy": new_action.ref, "prov:wasDerivedFrom": stored_results @@ -158,11 +170,12 @@ def __call__(self, args: argparse.Namespace) -> None: if merged_any: # One pass must have been completed already. new_action = prov_doc.add_activity(data={ + "schema:description": "merges the old data object with the new data", "prov:used": [last_data.ref, new_data.ref], "prov:wasInformedBy": [last_action.ref, new_action.ref], "prov:wasAssociatedWith": process_command.ref }) # initial merge action of the merge - merged_doc.prov_objects = [new_action, new_data, last_data] + merged_doc.prov_objects = [new_action, new_data, last_data] # set the starting objects of the merge self.log.info(f"### Merge data from {harvester} plugin") # merge data into the merge dict @@ -186,32 +199,55 @@ def __call__(self, args: argparse.Namespace) -> None: self.log.info("## Store processed metadata") # store processed data ctx.prepare_step("process") + begin_store_at_time = datetime.datetime.now().isoformat() with ctx["result"] as result_ctx: result_ctx["codemeta"] = merged_doc.compact() result_ctx["context"] = {"@context": merged_doc.full_context} result_ctx["expanded"] = merged_doc.ld_value + stored_at_time = datetime.datetime.now().isoformat() if prov_doc is not None: write = prov_doc.add_activity(data={ + "schema:description": "Writes the processed metadata into the HERMES cache.", "prov:wasAssociatedWith": [process_command.ref, hermes_cache.ref, plugin.ref], "prov:used": last_data.ref, - "prov:wasInformedBy": last_action.ref + "prov:wasInformedBy": last_action.ref, + "prov:startedAtTime": begin_store_at_time, + "prov:endedAtTime": stored_at_time }) # TODO: add more info prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The compacted version of the processed metadata.", + "schema:text": str(merged_doc.compact()), # TODO: maybe "prov:value" instead? + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "process" / "result" / "codemeta.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": last_data.ref, - "prov:wasAttributedTo": hermes_cache.ref + "prov:wasAttributedTo": hermes_cache.ref, + "prov:generatedAtTime": stored_at_time }) prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The context of the processed metadata.", + "schema:text": str({"@context": merged_doc.full_context}), # TODO: maybe "prov:value" instead? + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "process" / "result" / "context.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": last_data.ref, - "prov:wasAttributedTo": hermes_cache.ref + "prov:wasAttributedTo": hermes_cache.ref, + "prov:generatedAtTime": stored_at_time }) prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The expanded version of the processed metadata.", + "schema:text": str(merged_doc.ld_value), # TODO: maybe "prov:value" instead? + "schema:encodingFormat": "application/json", + "schema:url": (ctx.cache_dir / "process" / "result" / "expanded.json").absolute().as_uri(), "prov:wasGeneratedBy": write.ref, "prov:wasDerivedFrom": last_data.ref, - "prov:wasAttributedTo": hermes_cache.ref + "prov:wasAttributedTo": hermes_cache.ref, + "prov:generatedAtTime": stored_at_time }) with ctx["provenance"] as cache: diff --git a/src/hermes/model/merge/container.py b/src/hermes/model/merge/container.py index a286c7e8..470d3161 100644 --- a/src/hermes/model/merge/container.py +++ b/src/hermes/model/merge/container.py @@ -245,9 +245,10 @@ def __setitem__(self: Self, key: str, value: Union[JSON_LD_VALUE, BASIC_TYPE, TI create_new_merged_data = last_merged_data is self.prov_objects[2] elif self.prov_objects[0] is not None: merge_activity = self.prov_doc.add_activity(data={ - "schema:name": "merge", - "schema:description": "foo", - "prov:used": {"@list": [self.prov_objects[2].ref, str(self.path+[key])]}, + "schema:name": f"merge values at {str(self.path+[key])}", + "schema:description": f"inserting value in the second 'used' value at {str(self.path+[key])} into the " + "first 'used' value at the same point", + "prov:used": {"@list": [self.prov_objects[2].ref, self.prov_objects[1].ref]}, "prov:wasInformedBy": self.prov_objects[0].ref }) create_new_merged_data = True @@ -257,7 +258,13 @@ def __setitem__(self: Self, key: str, value: Union[JSON_LD_VALUE, BASIC_TYPE, TI return self.prov_objects[0] = merge_activity if create_new_merged_data: + outer_most_parent = self + while outer_most_parent.parent != None: + outer_most_parent = outer_most_parent.parent self.prov_objects[2] = self.prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": f"software metadata after merge of values at {str(self.path+[key])}", + "schema:text": str(outer_most_parent.compact()), # TODO: maybe "prov:value" instead? "prov:wasAttributedTo": self.prov_doc.get_hermes_command("process").ref, "prov:wasGeneratedBy": merge_activity.ref, "prov:wasDerivedFrom": {"@list": [self.prov_objects[1].ref, self.prov_objects[2].ref]} @@ -319,10 +326,15 @@ def _merge_item( raise MergeError(f"Can't merge, no strategy found for key '{key}'.") if self.prov_objects[0] is not None: merge_activity = self.prov_doc.add_activity(data={ - "schema:name": "merge", - "schema:description": "foo", + "schema:name": f"merge values at {str(self.path+[key])}", + "schema:description": f"merge value in the second 'used' value at {str(self.path+[key])} into the " + "first 'used' value at the same point using the third 'used' value", "prov:wasAssociatedWith": self.prov_doc.get_hermes_command("process").ref, - "prov:used": {"@list": [self.prov_objects[1].ref, self.prov_objects[2].ref, str(self.path+[key])]}, + "prov:used": {"@list": [ + self.prov_objects[2].ref, + self.prov_objects[1].ref, + f"{merger.merge.__module__}.{merger.merge.__qualname__}" + ]}, "prov:wasInformedBy": self.prov_objects[0].ref }) self.prov_objects[0] = merge_activity