diff --git a/src/hermes/commands/harvest/base.py b/src/hermes/commands/harvest/base.py index 0d3d9e5f..56799e9c 100644 --- a/src/hermes/commands/harvest/base.py +++ b/src/hermes/commands/harvest/base.py @@ -5,6 +5,9 @@ # SPDX-FileContributor: Michael Meinel import argparse +import datetime +from io import IOBase +from pathlib import Path from pydantic import BaseModel @@ -12,6 +15,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 class HermesHarvestPlugin(HermesPlugin): @@ -19,10 +23,36 @@ 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(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 + pass + class HarvestSettings(BaseModel): """Generic harvesting settings.""" @@ -30,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. """ @@ -37,9 +93,12 @@ 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 = self.init_provenance_document() + base_plugin = prov_doc.get_hermes_base_plugin("harvest") + 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.") @@ -62,17 +121,117 @@ 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 + 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 + outputs = [] + io_ops = [] + for plugin_io_operation in plugin_io_operations: + 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 = prov_doc.add_entity(data=plugin_io_operation[2]) + outputs.append(loaded_data.ref) + 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: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: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:startedAtTime": begin_store_at_time, + "prov:endedAtTime": stored_at_time + }) + prov_doc.add_entity(data={ + "@type": "schema:CreativeWork", + "schema:description": "The compacted version of the harvested metadata.", + "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, + "prov:wasDerivedFrom": data_output.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}), # 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, + "prov:wasDerivedFrom": data_output.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), # 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, + "prov:wasDerivedFrom": data_output.ref, + "prov:wasAttributedTo": prov_doc.get_hermes_cache().ref, + "prov:generatedAtTime": stored_at_time + }) + + with ctx["provenance"] as cache: + cache["result"] = prov_doc.ld_value + ctx.finalize_step('harvest') if not harvested_any: self.log.critical("No harvest plugin ran successfully.") raise HermesPluginRunError("No harvest plugin ran successfully.") + + def init_provenance_document(self) -> 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 KeyError: + pass + prov_doc = ld_prov_list() + prov_doc.init_hermes_agents() + return prov_doc 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) diff --git a/src/hermes/commands/process/base.py b/src/hermes/commands/process/base.py index 725f6487..b7d83a79 100644 --- a/src/hermes/commands/process/base.py +++ b/src/hermes/commands/process/base.py @@ -5,22 +5,25 @@ # SPDX-FileContributor: Michael Meinel import argparse -from typing import Union +import datetime +from typing import Optional 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 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,8 +41,14 @@ 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([{}]) + merged_doc = ld_merge_dict([{}], prov_doc) if not self.settings.plugins: self.log.critical( @@ -56,6 +65,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 +89,35 @@ 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={ + "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={ # 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 + }) + if not any_strategies_loaded: self.log.critical("## No process plugin was ran successfully.") raise HermesPluginRunError("No process plugin was ran successfully.") @@ -96,18 +135,60 @@ 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 + "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 + }) + 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] # set the starting objects of the merge + 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 + + 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 @@ -118,10 +199,73 @@ 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: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: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: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:generatedAtTime": stored_at_time + }) + + 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/merge/container.py b/src/hermes/model/merge/container.py index e9cc03a2..470d3161 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,39 @@ 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": 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 # 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: + 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]} + }) + else: + self.prov_objects[2]["prov:wasGeneratedBy"].append(merge_activity.ref) def match( self: Self, @@ -281,7 +324,23 @@ 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": 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[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 + 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 new file mode 100644 index 00000000..767d58ad --- /dev/null +++ b/src/hermes/model/provenance/ld_prov.py @@ -0,0 +1,159 @@ +# 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 + +from hermes import utils +from hermes.model.types import ld_dict, ld_list +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_list(ld_list): + 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, + 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__(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"], 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 + 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: + 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", "schema:Action"] + else: + activity["@type"].extend(["prov:Activity", "schema:Action"]) + 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", "schema:SoftwareApplication"] + else: + agent["@type"].extend(["prov:Agent", "schema:SoftwareApplication"]) + 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", "schema:Thing"] + else: + entity["@type"].extend(["prov:Entity", "schema:Thing"]) + 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": [*set(utils.hermes_urls.values())] + }) + 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 + }) + 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 = 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.get_hermes_base_plugin(step).ref + }) + return node + + def shallow_search(self, query) -> list[ld_dict]: + return [item for item in self if query(item)] + + def get_hermes(self) -> ld_dict: + 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 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 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 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 node: ( + "@id" in node and node["@id"] == ld_prov_list.HERMES_COMMAND_ID_FORMAT.format(step=step) + ))[0] 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 14331472..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: """ @@ -589,7 +630,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