Feature/454 e2e curate

src/hermes/commands/__init__.py

@@ -8,12 +8,12 @@
 # "unused import" errors.
 # flake8: noqa
 
-from hermes.commands.base import HermesHelpCommand
-from hermes.commands.base import HermesVersionCommand
-from hermes.commands.clean.base import HermesCleanCommand
-from hermes.commands.init.base import HermesInitCommand
+# from hermes.commands.base import HermesHelpCommand
+# from hermes.commands.base import HermesVersionCommand
+# from hermes.commands.clean.base import HermesCleanCommand
+# from hermes.commands.init.base import HermesInitCommand
 from hermes.commands.curate.base import HermesCurateCommand
 from hermes.commands.harvest.base import HermesHarvestCommand
-from hermes.commands.process.base import HermesProcessCommand
-from hermes.commands.deposit.base import HermesDepositCommand
-from hermes.commands.postprocess.base import HermesPostprocessCommand
+# from hermes.commands.process.base import HermesProcessCommand
+# from hermes.commands.deposit.base import HermesDepositCommand
+# from hermes.commands.postprocess.base import HermesPostprocessCommand

src/hermes/commands/base.py

@@ -9,19 +9,19 @@
 import logging
 import pathlib
 from importlib import metadata
-from typing import Dict, Optional, Type
+from typing import Type, Union
 
 import toml
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
-class _HermesSettings(BaseSettings):
+class HermesSettings(BaseSettings):
     """Root class for HERMES configuration model."""
 
     model_config = SettingsConfigDict(env_file_encoding='utf-8')
 
-    logging: Dict = {}
+    logging: dict = {}
 
 
 class HermesCommand(abc.ABC):
@@ -31,7 +31,7 @@ class HermesCommand(abc.ABC):
     """
 
     command_name: str = ""
-    settings_class: Type = _HermesSettings
+    settings_class: Type = HermesSettings
 
     def __init__(self, parser: argparse.ArgumentParser):
         """Initialize a new instance of any HERMES command.
@@ -45,28 +45,27 @@ def __init__(self, parser: argparse.ArgumentParser):
         self.log = logging.getLogger(f"hermes.{self.command_name}")
         self.errors = []
 
-    @classmethod
-    def init_plugins(cls):
+    def init_plugins(self):
         """Collect and initialize the plugins available for the HERMES command."""
 
         # Collect all entry points for this group (i.e., all valid plug-ins for the step)
-        entry_point_group = f"hermes.{cls.command_name}"
-        group_plugins = {
-            entry_point.name: entry_point.load()
-            for entry_point in metadata.entry_points(group=entry_point_group)
-        }
-
-        # Collect the plug-in specific configurations
-        cls.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
-        })
+        entry_point_group = f"hermes.{self.command_name}"
+        group_plugins = {}
+        group_settings = {}
+
+        for entry_point in metadata.entry_points(group=entry_point_group):
+            plugin_cls = entry_point.load()
+
+            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
 
     @classmethod
-    def derive_settings_class(cls, setting_types: Dict[str, Type]) -> None:
+    def derive_settings_class(cls, setting_types: dict[str, Type]) -> None:
         """Build a new Pydantic data model class for configuration.
 
         This will create a new class that includes all settings from the plugins available.
@@ -131,13 +130,10 @@ def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
 
     def load_settings(self, args: argparse.Namespace):
         """Load settings from the configuration file (passed in from command line)."""
-        try:
-            toml_data = toml.load(args.path / args.config)
-            self.root_settings = HermesCommand.settings_class.model_validate(toml_data)
-            self.settings = getattr(self.root_settings, self.command_name)
-        except FileNotFoundError as e:
-            self.log.error("hermes.toml was not found. Try to run 'hermes init' first or create one manually.")
-            raise e  # This will lead to our default error message & sys.exit
+
+        toml_data = toml.load(args.path / args.config)
+        self.root_settings = HermesCommand.settings_class.model_validate(toml_data)
+        self.settings = getattr(self.root_settings, self.command_name)
 
     def patch_settings(self, args: argparse.Namespace):
         """Process command line options for the settings."""
@@ -164,7 +160,9 @@ def __call__(self, args: argparse.Namespace):
 class HermesPlugin(abc.ABC):
     """Base class for all HERMES plugins."""
 
-    settings_class: Optional[Type] = None
+    pluing_node = None
+
+    settings_class: Union[Type, None] = None
 
     @abc.abstractmethod
     def __call__(self, command: HermesCommand) -> None:
@@ -202,27 +200,3 @@ def __call__(self, args: argparse.Namespace) -> None:
             # Otherwise, simply show the general help and exit (cleanly).
             self.parser.print_help()
             self.parser.exit()
-
-    def load_settings(self, args: argparse.Namespace):
-        """No settings are needed for the help command."""
-        pass
-
-
-class HermesVersionSettings(BaseModel):
-    """Intentionally empty settings class for the version command."""
-    pass
-
-
-class HermesVersionCommand(HermesCommand):
-    """Show HERMES version and exit."""
-
-    command_name = "version"
-    settings_class = HermesVersionSettings
-
-    def load_settings(self, args: argparse.Namespace):
-        """Pass loading settings as not necessary for this command."""
-        pass
-
-    def __call__(self, args: argparse.Namespace) -> None:
-        self.log.info(metadata.version("hermes"))
-        self.parser.exit()

src/hermes/commands/cli.py

@@ -12,9 +12,11 @@
 import sys
 
 from hermes import logger
-from hermes.commands import (HermesHelpCommand, HermesVersionCommand, HermesCleanCommand,
-                             HermesHarvestCommand, HermesProcessCommand, HermesCurateCommand,
-                             HermesDepositCommand, HermesPostprocessCommand, HermesInitCommand)
+# FIXME: remove comments after new implementation of modules is available
+# from hermes.commands import (HermesHelpCommand, HermesVersionCommand, HermesCleanCommand,
+#                              HermesHarvestCommand, HermesProcessCommand, HermesCurateCommand,
+#                              HermesDepositCommand, HermesPostprocessCommand, HermesInitCommand)
+from hermes.commands import HermesCurateCommand, HermesHarvestCommand
 from hermes.commands.base import HermesCommand
 
 
@@ -36,15 +38,15 @@ def main() -> None:
     setting_types = {}
 
     for command in (
-            HermesHelpCommand(parser),
-            HermesVersionCommand(parser),
-            HermesInitCommand(parser),
-            HermesCleanCommand(parser),
+            # HermesHelpCommand(parser),
+            # HermesVersionCommand(parser),
+            # HermesInitCommand(parser),
+            # HermesCleanCommand(parser),
             HermesHarvestCommand(parser),
-            HermesProcessCommand(parser),
+            # HermesProcessCommand(parser),
             HermesCurateCommand(parser),
-            HermesDepositCommand(parser),
-            HermesPostprocessCommand(parser),
+            # HermesDepositCommand(parser),
+            # HermesPostprocessCommand(parser),
     ):
         if command.settings_class is not None:
             setting_types[command.command_name] = command.settings_class

src/hermes/commands/curate/base.py

@@ -5,17 +5,16 @@
 # SPDX-FileContributor: Michael Meinel
 
 import argparse
-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 import SoftwareMetadata
+from hermes.model.context_manager import HermesContext
+from hermes.model.error import HermesValidationError
 
 
-class _CurateSettings(BaseModel):
+class CurateSettings(BaseModel):
     """Generic deposition settings."""
 
     pass
@@ -25,23 +24,30 @@ class HermesCurateCommand(HermesCommand):
     """ Curate the unified metadata before deposition. """
 
     command_name = "curate"
-    settings_class = _CurateSettings
+    settings_class = CurateSettings
 
     def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
         pass
 
     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")
+
+        ctx.prepare_step("process")
+        with ctx["result"] as process_ctx:
+            expanded_data = process_ctx["expanded"]
+            context_data = process_ctx["context"]
+        ctx.finalize_step("process")
+
+        try:
+            data = SoftwareMetadata(expanded_data[0], context_data["@context"][1])
+        except Exception as e:
+            raise HermesValidationError("The results of the process step are invalid.") from e
 
-        if not process_output.is_file():
-            self.log.error(
-                "No processed metadata found. Please run `hermes process` before curation."
-            )
-            sys.exit(1)
+        with ctx["result"] as curate_ctx:
+            curate_ctx["expanded"] = data.ld_value
+            curate_ctx["context"] = {"@context": data.full_context}
 
-        os.makedirs(ctx.hermes_dir / 'curate', exist_ok=True)
-        shutil.copy(process_output, ctx.hermes_dir / 'curate' / (ctx.hermes_name + '.json'))
+        ctx.finalize_step("curate")

src/hermes/commands/harvest/base.py

@@ -5,14 +5,13 @@
 # SPDX-FileContributor: Michael Meinel
 
 import argparse
-import typing as t
-from datetime import datetime
 
 from pydantic import BaseModel
 
 from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import HermesContext, HermesHarvestContext
-from hermes.model.error import HermesValidationError, HermesMergeError
+from hermes.model.context_manager import HermesContext
+from hermes.model.error import HermesValidationError
+from hermes.model import SoftwareMetadata
 
 
 class HermesHarvestPlugin(HermesPlugin):
@@ -21,11 +20,11 @@ class HermesHarvestPlugin(HermesPlugin):
     TODO: describe the harvesting process and how this is mapped to this plugin.
     """
 
-    def __call__(self, command: HermesCommand) -> t.Tuple[t.Dict, t.Dict]:
+    def __call__(self, command: HermesCommand) -> tuple[SoftwareMetadata, dict]:
         pass
 
 
-class _HarvestSettings(BaseModel):
+class HarvestSettings(BaseModel):
     """Generic harvesting settings."""
 
     sources: list[str] = []
@@ -35,32 +34,31 @@ class HermesHarvestCommand(HermesCommand):
     """ Harvest metadata from configured sources. """
 
     command_name = "harvest"
-    settings_class = _HarvestSettings
+    settings_class = HarvestSettings
 
     def __call__(self, args: argparse.Namespace) -> None:
         self.args = args
-        ctx = HermesContext()
 
         # Initialize the harvest cache directory here to indicate the step ran
-        ctx.init_cache("harvest")
+        ctx = HermesContext()
+        ctx.prepare_step('harvest')
 
         for plugin_name in self.settings.sources:
+            plugin_cls = self.plugins[plugin_name]
+
             try:
-                plugin_func = self.plugins[plugin_name]()
-                harvested_data, tags = plugin_func(self)
-
-                with HermesHarvestContext(ctx, plugin_name) as harvest_ctx:
-                    harvest_ctx.update_from(harvested_data,
-                                            plugin=plugin_name,
-                                            timestamp=datetime.now().isoformat(), **tags)
-                    for _key, ((_value, _tag), *_trace) in harvest_ctx._data.items():
-                        if any(v != _value and t == _tag for v, t in _trace):
-                            raise HermesMergeError(_key, None, _value)
-
-            except KeyError as e:
-                self.log.error("Plugin '%s' not found.", plugin_name)
-                self.errors.append(e)
+                # Load plugin and run the harvester
+                plugin_func = plugin_cls()
+                harvested_data = plugin_func(self)
+
+                with ctx[plugin_name] as plugin_ctx:
+                    plugin_ctx["codemeta"] = harvested_data[0].compact()
+                    plugin_ctx["context"] = {"@context": harvested_data[0].full_context}
+
+                    plugin_ctx["expanded"] = harvested_data[0].ld_value
 
             except HermesValidationError as e:
                 self.log.error("Error while executing %s: %s", plugin_name, e)
                 self.errors.append(e)
+
+        ctx.finalize_step('harvest')