Add the capability to upgrade a .ttl file version using AdaptiveGraph

Author: Oleksandr Muzyka (EPAM)

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/adaptive_graph.py

@@ -0,0 +1,144 @@
+import pathlib
+import subprocess
+import tempfile
+
+from typing import Optional, Union
+
+from rdflib import Graph
+
+from esmf_aspect_meta_model_python.constants import SAMM_NAMESPACE_PREFIX, SAMM_ORG_IDENTIFIER
+from esmf_aspect_meta_model_python.samm_cli import SammCli
+
+
+class AdaptiveGraph(Graph):  # TODO: avoid double parse when upgrading
+    """An RDF graph that can adaptively upgrade SAMM files using the SAMM CLI."""
+
+    def __init__(self, samm_version: str | None = None, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+        self._samm_version = samm_version
+        self._samm_cli = SammCli()
+
+    def _upgrade_ttl_file(self, file_path: pathlib.Path) -> str:
+        """Run SAMM CLI prettyprint to upgrade a TTL file to the latest version."""
+        try:
+            return self._samm_cli.prettyprint(str(file_path), capture=True)
+        except subprocess.CalledProcessError as e:
+            raise RuntimeError(f"SAMM CLI failed for {file_path}:\n{e.stdout}\n{e.stderr}") from e
+
+    def _has_version_mismatch_in_graph(self, input_source: str | bytes | pathlib.PurePath) -> bool:
+        """Parse into a temporary graph to detect mismatch BEFORE loading into self."""
+        temp_graph = Graph()
+
+        if isinstance(input_source, pathlib.PurePath):
+            temp_graph.parse(input_source)
+        else:  # str | bytes
+            temp_graph.parse(data=input_source, format="ttl")
+
+        for prefix, namespace in temp_graph.namespace_manager.namespaces():
+            if prefix.startswith(SAMM_NAMESPACE_PREFIX):
+                parts = namespace.strip("#").split(":")
+                if len(parts) >= 5 and parts[2] == SAMM_ORG_IDENTIFIER:
+                    version = parts[-1]
+                    if version != self._samm_version:
+                        return True
+        return False
+
+    def _upgrade_source(self, source: str | pathlib.PurePath) -> str:
+        source_path = pathlib.Path(source)
+
+        print(f"[INFO] SAMM version mismatch detected in {source_path}. Upgrading...")
+
+        return self._upgrade_ttl_file(source_path)
+
+    def _upgrade_data(self, data: str | bytes) -> str:
+        print(  # TODO: improve logging
+            f"[INFO] SAMM version mismatch detected in provided data (target v{self._samm_version}) Upgrading..."
+        )
+
+        with tempfile.NamedTemporaryFile("wb", suffix=".ttl", delete=False) as tmp:
+            tmp.write(data.encode("utf-8") if isinstance(data, str) else data)
+            tmp_path = pathlib.Path(tmp.name)
+
+        try:
+            return self._upgrade_ttl_file(tmp_path)
+        finally:
+            tmp_path.unlink(missing_ok=True)
+
+    def set_samm_version(self, samm_version: str | None) -> None:
+        """Set the SAMM version for this graph."""
+        self._samm_version = samm_version
+
+    def parse(  # type: ignore[override]
+        self,
+        *,
+        source: Optional[pathlib.PurePath] = None,
+        data: Optional[str | bytes] = None,
+        **kwargs,
+    ) -> "AdaptiveGraph":
+        """
+        Parse a TTL file into this graph, upgrading via SAMM CLI if version mismatch detected.
+
+        If a SAMM version mismatch is detected, the TTL file will be upgraded using the SAMM CLI prettyprint
+        before parsing into this graph.
+
+        Args:
+            source: Path to the TTL file as pathlib.Path.
+            data: RDF content as string or bytes.
+            **kwargs: Additional arguments passed to rdflib.Graph.parse().
+
+        Returns:
+            self (AdaptiveGraph): The current graph instance with parsed data.
+
+        Raises:
+            RuntimeError: If the SAMM CLI fails during the upgrade process.
+            ValueError: If neither 'source' nor 'data' is provided, or if SAMM version is not set.
+        """
+        if (source is None) == (data is None):
+            raise ValueError("Either 'source' or 'data' must be provided.")
+        if not self._samm_version:
+            raise ValueError("SAMM version is not set.")
+
+        if self._has_version_mismatch_in_graph(input_source=data or pathlib.Path(source)):  # type: ignore[arg-type]
+            data = self._upgrade_data(data) if data else self._upgrade_source(source)  # type: ignore[arg-type]
+            source = None
+
+        super().parse(source=source, data=data, **kwargs)
+
+        return self
+
+    def __add__(self, other: Union["Graph", "AdaptiveGraph"]) -> "AdaptiveGraph":
+        """Override addition to propagate SAMM version to the resulting graph."""
+        retval = super().__add__(other)
+
+        if isinstance(retval, AdaptiveGraph):
+            if isinstance(other, AdaptiveGraph) and other._samm_version != self._samm_version:
+                raise ValueError("SAMM version mismatch during addition.")
+
+            retval.set_samm_version(self._samm_version)
+
+        return retval  # type: ignore[return-value]
+
+    def __sub__(self, other: Union["Graph", "AdaptiveGraph"]) -> "AdaptiveGraph":
+        """Override subtraction to propagate SAMM version to the resulting graph."""
+        retval = super().__sub__(other)
+
+        if isinstance(retval, AdaptiveGraph):
+            if isinstance(other, AdaptiveGraph) and other._samm_version != self._samm_version:
+                raise ValueError("SAMM version mismatch during subtraction.")
+
+            retval.set_samm_version(self._samm_version)
+
+        return retval  # type: ignore[return-value]
+
+    def __mul__(self, other: Union["Graph", "AdaptiveGraph"]) -> "AdaptiveGraph":
+        """Override multiplication to propagate SAMM version to the resulting graph."""
+        retval = super().__mul__(other)
+
+        if isinstance(retval, AdaptiveGraph):
+            if isinstance(other, AdaptiveGraph) and other._samm_version != self._samm_version:
+                raise ValueError("SAMM version mismatch during multiplication.")
+
+            retval.set_samm_version(self._samm_version)
+
+        return retval  # type: ignore[return-value]

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/constants.py

@@ -12,3 +12,6 @@
 
 SAMM_VERSION = "2.2.0"
 JAVA_CLI_VERSION = "2.11.1"
+
+SAMM_NAMESPACE_PREFIX = "samm"
+SAMM_ORG_IDENTIFIER = "org.eclipse.esmf.samm"

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/loader/samm_graph.py

@@ -14,9 +14,11 @@
 
 from rdflib import RDF, Graph, Node
 
+from esmf_aspect_meta_model_python.adaptive_graph import AdaptiveGraph
 from esmf_aspect_meta_model_python.base.aspect import Aspect
 from esmf_aspect_meta_model_python.base.base import Base
 from esmf_aspect_meta_model_python.base.property import Property
+from esmf_aspect_meta_model_python.constants import SAMM_NAMESPACE_PREFIX, SAMM_ORG_IDENTIFIER
 from esmf_aspect_meta_model_python.impl.base_impl import BaseImpl
 from esmf_aspect_meta_model_python.loader.default_element_cache import DefaultElementCache
 from esmf_aspect_meta_model_python.loader.model_element_factory import ModelElementFactory
@@ -28,10 +30,8 @@
 class SAMMGraph:
     """Class representing the SAMM graph and its operations."""
 
-    samm_namespace_prefix = "samm"
-
     def __init__(self):
-        self.rdf_graph = Graph()
+        self.rdf_graph = AdaptiveGraph()
         self.samm_graph = Graph()
         self._cache = DefaultElementCache()
 
@@ -59,7 +59,7 @@ def _get_rdf_graph(self, input_data: Union[str, Path], input_type: Optional[str]
         """Read the RDF graph from the given input data.
 
         This method initializes the `InputHandler` with the provided input data and type,
-        retrieves the reader, and reads the RDF graph into `self.rdf_graph`.
+        retrieves the reader, sets the SAMM version, and reads the RDF graph into `self.rdf_graph`.
 
         Args:
             input_data (Union[str, Path]): The input data to read the RDF graph from. This can be a file path or a str.
@@ -69,26 +69,9 @@ def _get_rdf_graph(self, input_data: Union[str, Path], input_type: Optional[str]
             None
         """
         self._reader = InputHandler(input_data, input_type).get_reader()
+        self._reader.set_samm_version(input_data)
         self.rdf_graph = self._reader.read(input_data)
 
-    def _get_samm_version_from_rdf_graph(self) -> str:
-        """Extracts the SAMM version from the RDF graph.
-
-        This method searches through the RDF graph namespaces to find a prefix that indicates the SAMM version.
-
-        Returns:
-            str: The SAMM version as a string extracted from the graph. Returns an empty string if no version
-                 can be conclusively identified.
-        """
-        version = ""
-
-        for prefix, namespace in self.rdf_graph.namespace_manager.namespaces():
-            if prefix == self.samm_namespace_prefix:
-                urn_parts = namespace.split(":")
-                version = urn_parts[-1].replace("#", "")
-
-        return version
-
     def _get_samm_version(self):
         """Retrieve and set the SAMM version from the RDF graph.
 
@@ -98,7 +81,7 @@ def _get_samm_version(self):
         Raises:
             ValueError: If the SAMM version is not found in the RDF graph.
         """
-        self.samm_version = self._get_samm_version_from_rdf_graph()
+        self.samm_version = self._reader.samm_version
 
         if not self.samm_version:
             raise ValueError(
@@ -220,7 +203,7 @@ def load_aspect_model(self) -> Aspect:
 
         return self.aspect
 
-    def _validate_samm_namespace_version(self, graph: Graph) -> None:
+    def _validate_samm_namespace_version(self, graph: AdaptiveGraph) -> None:
         """
         Validates that the SAMM version in the given RDF graph matches the detected SAMM version.
 
@@ -230,16 +213,16 @@ def _validate_samm_namespace_version(self, graph: Graph) -> None:
         do not match, a `ValueError` is raised.
 
         Args:
-            graph (Graph): The RDF graph whose namespaces are to be validated.
+            graph (AdaptiveGraph): The RDF graph whose namespaces are to be validated.
 
         Raises:
             ValueError: If the SAMM version in the graph's namespace does not match the detected SAMM version.
         """
         for prefix, namespace in graph.namespace_manager.namespaces():
-            if prefix.startswith(self.samm_namespace_prefix):
+            if prefix.startswith(SAMM_NAMESPACE_PREFIX):
                 namespace_info = namespace.split(":")  # [urn, namespace_id, namespace_specific_str, entity, version#]
 
-                if len(namespace_info) == 5 and namespace_info[2] == "org.eclipse.esmf.samm":
+                if len(namespace_info) == 5 and namespace_info[2] == SAMM_ORG_IDENTIFIER:
                     version = namespace_info[-1].replace("#", "")
 
                     if version != self.samm_version:

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/resolver/base.py

@@ -15,6 +15,8 @@
 
 from rdflib import Graph
 
+from esmf_aspect_meta_model_python.adaptive_graph import AdaptiveGraph
+from esmf_aspect_meta_model_python.constants import SAMM_NAMESPACE_PREFIX
 from esmf_aspect_meta_model_python.samm_meta_model import SammUnitsGraph
 
 
@@ -33,7 +35,7 @@ class ResolverInterface(ABC):
     """
 
     def __init__(self):
-        self.graph = Graph()
+        self.graph = AdaptiveGraph()
         self.samm_graph = None
         self.samm_version = ""
 
@@ -70,46 +72,47 @@ def _validate_samm_version(samm_version: str):
         elif samm_version > SammUnitsGraph.SAMM_VERSION:
             raise ValueError(f"{samm_version} is not supported SAMM version.")
 
-    def _get_samm_version_from_graph(self) -> str:
+    def set_samm_version(self, steam_input: Union[str, Path]) -> None:
         """
-        Extracts the SAMM version from the RDF graph.
+        Sets the SAMM version by extracting it from the specified file.
 
-        This method searches through the RDF graph namespaces to find a prefix that indicate the SAMM version.
+        This method uses the AdaptiveGraph class to extract the SAMM version from the given file.
+        There is also a validation against known SAMM versions to ensure the version is supported and recognized.
 
-        Returns:
-            str: The SAMM version as a string extracted from the graph. Returns an empty string if no version
-                 can be conclusively identified.
-        """
-        version = ""
-
-        for prefix, namespace in self.graph.namespace_manager.namespaces():
-            if prefix == "samm":
-                urn_parts = namespace.split(":")
-                version = urn_parts[-1].replace("#", "")
-
-        return version
+        Args:
+            steam_input (Union[str, Path]): The path to the file from which the SAMM version is to be extracted.
 
-    def get_samm_version(self) -> str:
+        Raises:
+            ValueError: If the extracted version is not supported or if it is not found in the file.
         """
-        Retrieves and validates the specified SAMM version from the provided Aspect model graph.
+        version = self.get_samm_version_from_input(steam_input)
+        self._validate_samm_version(version)
+        self.samm_version = version
 
-        This method attempts to extract the version information of the SAMM from a graph. There is also a validation
-        against known SAMM versions to ensure the version is supported and recognized.
+    def get_samm_version_from_input(self, stream_input: Union[str, Path]) -> str:
+        """
+        Retrieves the latest SAMM version from the specified stream input.
 
+        Args:
+            stream_input (Union[str, Path]): The path to the file from which the SAMM version is to be extracted.
 
         Returns:
-            str: The validated version of SAMM if it is recognized and supported. If the version is not valid,
-                 an appropriate message or value indicating non-recognition is returned.
+            str: The extracted SAMM version.
+        """
+        temp_graph = Graph()
+        version = ""
 
-        Raises:
-            ValueError: If the extracted version is not supported or if it is not found in the Graph.
+        if isinstance(stream_input, Path):
+            temp_graph.parse(source=stream_input, format="turtle")
+        else:
+            temp_graph.parse(data=stream_input)
 
-        """
-        version = self._get_samm_version_from_graph()
-        self._validate_samm_version(version)
-        self.samm_version = version
+        for prefix, namespace in temp_graph.namespace_manager.namespaces():
+            if prefix.startswith(SAMM_NAMESPACE_PREFIX):
+                urn_parts = namespace.split(":")
+                version = urn_parts[-1].replace("#", "")
 
         return version
 
-    def prepare_aspect_model(self, graph: Graph):
+    def prepare_aspect_model(self, graph: AdaptiveGraph):
         """Resolve all additional graph elements if needed."""

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/resolver/data_string.py

@@ -12,8 +12,7 @@
 from pathlib import Path
 from typing import Union
 
-from rdflib import Graph
-
+from esmf_aspect_meta_model_python.adaptive_graph import AdaptiveGraph
 from esmf_aspect_meta_model_python.resolver.base import ResolverInterface
 
 
@@ -33,7 +32,7 @@ def read(self, data_string: Union[str, Path]):
         Returns:
             RDFGraph: An object representing the RDF graph constructed from the input data.
         """
-        self.graph = Graph()
+        self.graph = AdaptiveGraph(samm_version=self.samm_version)
         self.graph.parse(data=str(data_string) if isinstance(data_string, Path) else data_string)
 
         return self.graph