Update plugins to have a common class interface and to use the new data model.

Author: led02

src/hermes/commands/base.py

@@ -15,6 +15,10 @@
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from hermes import utils
+from hermes.model.prov.ld_prov import ld_prov, ld_prov_node
+from hermes.model.types import ld_context
+
 
 class HermesSettings(BaseSettings):
     """Root class for HERMES configuration model."""
@@ -33,15 +37,53 @@ class HermesCommand(abc.ABC):
     command_name: str = ""
     settings_class: Type = HermesSettings
 
+    class prov:
+        @classmethod
+        def hermes_software(cls):
+            return {
+                "@type": "schema:SoftwareApplication",
+                "schema:name": utils.hermes_name,
+                "schema:version": utils.hermes_version,
+                "schema:url": utils.hermes_homepage,
+            }
+
+        @classmethod
+        def hermes_command(cls, cmd, app):
+            return {
+                "@type": "schema:SoftwareApplication",
+                "schema:name": cmd.command_name,
+                "schema:isPartOf": app.ref,
+            }
+
+        @classmethod
+        def hermes_plugin_run(cls, plugin, command):
+            return {
+                "prov:wasStaredBy": command.ref,
+            }
+
+        @classmethod
+        def hermes_json_data(cls, name, data):
+            return {
+                "@type": "schema:PropertyValue",
+                "schema:name": name,
+                "schema:value": {"@type": "@json", "@value": data.compact()},
+            }
+
+
     def __init__(self, parser: argparse.ArgumentParser):
         """Initialize a new instance of any HERMES command.
 
         :param parser: The command line parser used for reading command line arguments.
         """
+        self.prov_doc = ld_prov(context=ld_context.ALL_CONTEXTS)
+        self.app_entity = self.prov_doc.make_node('Entity', self.prov.hermes_software())
+
         self.parser = parser
         self.plugins = self.init_plugins()
         self.settings = None
 
+        self.app_entity.commit()
+
         self.log = logging.getLogger(f"hermes.{self.command_name}")
         self.errors = []
 
@@ -50,17 +92,20 @@ def init_plugins(self):
 
         # Collect all entry points for this group (i.e., all valid plug-ins for the step)
         entry_point_group = f"hermes.{self.command_name}"
-        group_plugins = {
-            entry_point.name: entry_point.load()
-            for entry_point in metadata.entry_points(group=entry_point_group)
-        }
+        group_plugins = {}
+        group_settings = {}
+
+        for entry_point in metadata.entry_points(group=entry_point_group):
+            plugin_cls = entry_point.load()
+            ep_metadata = plugin_cls.get_metadata(entry_point)
+
+            plugin_cls.plugin_node = self.app_entity.add_related("schema:hasPart", "Entity", ep_metadata)
 
-        # Collect the plug-in specific configurations
-        self.derive_settings_class({
-            plugin_name: plugin_class.settings_class
-            for plugin_name, plugin_class in group_plugins.items()
-            if hasattr(plugin_class, "settings_class") and plugin_class.settings_class is not None
-        })
+            group_plugins[entry_point.name] = plugin_cls
+            if hasattr(plugin_cls, 'settings_class') and plugin_cls.settings_class is not None:
+                group_settings[entry_point.name] = plugin_cls.settings_class
+
+        self.derive_settings_class(group_settings)
 
         return group_plugins
 
@@ -160,8 +205,22 @@ def __call__(self, args: argparse.Namespace):
 class HermesPlugin(abc.ABC):
     """Base class for all HERMES plugins."""
 
+    pluing_node = None
+
     settings_class: Optional[Type] = None
 
+    @classmethod
+    def get_metadata(cls, entry_point):
+        cls.entry_point = entry_point
+
+        return {
+            "@type": "schema:EntryPoint",
+            "schema:name": entry_point.name,
+        }
+
+    def __init__(self, plugin_prov):
+        self.prov_doc = plugin_prov
+
     @abc.abstractmethod
     def __call__(self, command: HermesCommand) -> None:
         """Execute the plugin.

src/hermes/commands/clean/base.py

@@ -27,4 +27,6 @@ def __call__(self, args: argparse.Namespace) -> None:
         self.log.info("Removing HERMES caches...")
 
         # Naive implementation for now... check errors, validate directory, don't construct the path ourselves, etc.
-        shutil.rmtree(args.path / '.hermes')
+        cache_path = args.path / '.hermes'
+        if cache_path.exists():
+            shutil.rmtree(cache_path)

src/hermes/commands/cli.py

@@ -72,7 +72,7 @@ def main() -> None:
 
         log.info("Run subcommand %s", args.command.command_name)
         args.command(args)
-    except Exception as e:
+    except RuntimeError as e:
         log.error("An error occurred during execution of %s", args.command.command_name)
         log.debug("Original exception was: %s", e)
 

src/hermes/commands/curate/base.py

@@ -5,14 +5,16 @@
 # SPDX-FileContributor: Michael Meinel
 
 import argparse
+import json
 import os
 import shutil
 import sys
 
 from pydantic import BaseModel
 
 from hermes.commands.base import HermesCommand
-from hermes.model.context import CodeMetaContext
+from hermes.model.context_manager import HermesContext
+from hermes.model.types import ld_dict, ld_list
 
 
 class CurateSettings(BaseModel):
@@ -34,14 +36,33 @@ def __call__(self, args: argparse.Namespace) -> None:
 
         self.log.info("# Metadata curation")
 
-        ctx = CodeMetaContext()
-        process_output = ctx.hermes_dir / 'process' / (ctx.hermes_name + ".json")
+        ctx = HermesContext()
+        ctx.prepare_step("curate")
 
-        if not process_output.is_file():
-            self.log.error(
-                "No processed metadata found. Please run `hermes process` before curation."
-            )
-            sys.exit(1)
+        ctx.prepare_step("process")
+        with ctx["result"] as process_ctx:
+            expanded_data = process_ctx["expanded"]
+            context_data = process_ctx["context"]
+            prov_data = process_ctx["prov"]
+        ctx.finalize_step("process")
 
-        os.makedirs(ctx.hermes_dir / 'curate', exist_ok=True)
-        shutil.copy(process_output, ctx.hermes_dir / 'curate' / (ctx.hermes_name + '.json'))
+        prov_doc = ld_dict.from_dict({"hermes-rt:graph": prov_data, "@context": prov_data["@context"]})
+
+        nodes = {}
+        edges = {}
+
+        for node in prov_doc["hermes-rt:graph"]:
+            nodes[node["@id"]] = node
+
+            for rel in ('schema:isPartOf', "schema:hasPart", "prov:used", "prov:generated", "prov:wasStartedBy"):
+                if rel in node:
+                    rel_ids = node[rel]
+                    if not isinstance(rel_ids, ld_list):
+                        rel_ids = [rel_ids]
+                    edges[rel] = edges.get(rel, []) + [(node["@id"], rel_id) for rel_id in rel_ids]
+
+        with ctx["result"] as curate_ctx:
+            curate_ctx["expanded"] = expanded_data
+            curate_ctx["context"] = context_data
+
+        ctx.finalize_step("curate")

src/hermes/commands/deposit/base.py

@@ -7,14 +7,12 @@
 
 import abc
 import argparse
-import json
-import sys
 
 from pydantic import BaseModel
 
 from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import CodeMetaContext
-from hermes.model.path import ContextPath
+from hermes.model.context_manager import HermesContext
+from hermes.model.types import ld_dict
 from hermes.model.errors import HermesValidationError
 
 
@@ -24,16 +22,22 @@ class BaseDepositPlugin(HermesPlugin):
     TODO: describe workflow... needs refactoring to be less stateful!
     """
 
-    def __init__(self, command, ctx):
-        self.command = command
-        self.ctx = ctx
-
     def __call__(self, command: HermesCommand) -> None:
         """Initiate the deposition process.
 
         This calls a list of additional methods on the class, none of which need to be implemented.
         """
         self.command = command
+        ctx = HermesContext()
+        ctx.prepare_step("deposit")
+
+        ctx.prepare_step("curate")
+        with ctx["result"] as curate_ctx:
+            expanded_data = curate_ctx["expanded"]
+            context_data = curate_ctx["context"]
+        ctx.finalize_step("curate")
+
+        self.ctx = ld_dict(expanded_data, context=context_data)
 
         self.prepare()
         self.map_metadata()
@@ -128,26 +132,12 @@ def __call__(self, args: argparse.Namespace) -> None:
         self.args = args
         plugin_name = self.settings.target
 
-        ctx = CodeMetaContext()
-        codemeta_file = ctx.get_cache("curate", ctx.hermes_name)
-        if not codemeta_file.exists():
-            self.log.error("You must run the 'curate' command before deposit")
-            sys.exit(1)
-
-        codemeta_path = ContextPath("codemeta")
-        with open(codemeta_file) as codemeta_fh:
-            ctx.update(codemeta_path, json.load(codemeta_fh))
-
         try:
-            plugin_func = self.plugins[plugin_name](self, ctx)
-
+            plugin_func = self.plugins[plugin_name](self.prov_doc)
+            plugin_func(self)
         except KeyError as e:
             self.log.error("Plugin '%s' not found.", plugin_name)
             self.errors.append(e)
-
-        try:
-            plugin_func(self)
-
         except HermesValidationError as e:
             self.log.error("Error while executing %s: %s", plugin_name, e)
             self.errors.append(e)

src/hermes/commands/deposit/file.py

@@ -11,22 +11,34 @@
 from pydantic import BaseModel
 
 from hermes.commands.deposit.base import BaseDepositPlugin
-from hermes.model.path import ContextPath
+from hermes.model.types import ld_list, ld_dict, ld_context
 
 
 class FileDepositSettings(BaseModel):
-    filename: str = 'hermes.json'
+    filename: str = 'codemeta.json'
+    include_prov: bool = False
 
 
 class FileDepositPlugin(BaseDepositPlugin):
     settings_class = FileDepositSettings
 
     def map_metadata(self) -> None:
-        self.ctx.update(ContextPath.parse('deposit.file'), self.ctx['codemeta'])
+        if not self.command.settings.file.include_prov:
+            self._strip_namespace(self.ctx, ld_context.HERMES_RT_PREFIX)
+
+    def _strip_namespace(self, ctx, ns):
+        if isinstance(ctx, ld_dict):
+            for key, value in [*ctx.items()]:
+                if key.startswith(ns):
+                    del ctx[key]
+                else:
+                    self._strip_namespace(value, ns)
+        elif isinstance(ctx, ld_list):
+            for item in ctx:
+                self._strip_namespace(item, ns)
 
     def publish(self) -> None:
         file_config = self.command.settings.file
-        output_data = self.ctx['deposit.file']
 
         with open(file_config.filename, 'w') as deposition_file:
-            json.dump(output_data, deposition_file, indent=2)
+            json.dump(self.ctx.compact(), deposition_file, indent=2)