Add resolver for parsing graph from string

Author: Hanna-Shalamitskaya-EPAM

core/esmf-aspect-meta-model-python/README.md

@@ -9,6 +9,7 @@ https://projects.eclipse.org/projects/dt.esmf).
     * [Download SAMM files](#download-samm-files)
       * [Download SAMM release](#download-samm-release)
       * [Download SAMM branch](#download-samm-branch)
+  * [Input data handler usage](#input-data-handler-usage)
   * [Aspect Meta Model Loader usage](#aspect-meta-model-loader-usage)
   * [Samm Units](#samm-units)
   * [SAMM CLI wrapper class](#samm-cli-wrapper-class)
@@ -84,14 +85,47 @@ poetry run download-samm-branch
 ```
 Link to all branches: [SAMM Releases](https://github.com/eclipse-esmf/esmf-semantic-aspect-meta-model/branches)
 
+## Input data handler usage
+
+The InputHandler is a general-purpose class designed for loading input data into an RDF graph. 
+It easily accommodates different input sources such as local files (.ttl) or direct data strings containing 
+RDF formatted data.
+
+```python
+from esmf_aspect_meta_model_python.resolver.handler import InputHandler
+
+# Instantiating the Handler
+# The InputHandler requires a path or data string upon instantiation, which defines the source of RDF data
+# local file
+model_path = "path/to/local/file/AspectName.ttl"
+handler = InputHandler(model_path)
+graph, aspect_urn = handler.get_rdf_graph()
+
+# returns a tuple containing the RDF graph and the aspect URN derived from the provided data source
+```
+
+```python
+from esmf_aspect_meta_model_python.resolver.handler import InputHandler
+
+# Alternatively, if you have RDF data in a string format, you can directly pass it as follows:
+rdf_data_string = "your RDF data string here"
+handler = InputHandler(rdf_data_string)
+graph, aspect_urn = handler.get_rdf_graph()
+```
+
 ## Aspect Meta Model Loader usage
 
 An Aspect of the Meta Model can be loaded as follows:
 ```python
 from esmf_aspect_meta_model_python import AspectLoader
+from esmf_aspect_meta_model_python.resolver.handler import InputHandler
+
+model_path = "absolute/path/to/turtle.ttl"
+handler = InputHandler(model_path)
+graph, aspect_urn = handler.get_rdf_graph()
 
 loader = AspectLoader()
-model_elements = loader.load_aspect_model("absolute/path/to/turtle.ttl")
+model_elements = loader.load_aspect_model(graph, aspect_urn)
 aspect = model_elements[0]
 
 # or you can provide an Aspect URN

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

@@ -87,3 +87,4 @@
 )
 from .loader.aspect_loader import AspectLoader
 from .loader.samm_graph import SAMMGraph
+from .resolver.handler import InputHandler

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

@@ -9,10 +9,11 @@
 #
 #   SPDX-License-Identifier: MPL-2.0
 
-from pathlib import Path
-from typing import Union
+from typing import List
 
-from esmf_aspect_meta_model_python.loader.default_element_cache import DefaultElementCache
+from rdflib import Graph, URIRef
+
+from esmf_aspect_meta_model_python.base.aspect import Aspect
 from esmf_aspect_meta_model_python.loader.samm_graph import SAMMGraph
 
 
@@ -24,55 +25,37 @@ class AspectLoader:
     The default cache strategy ignores inline defined elements.
     """
 
-    def __init__(self) -> None:
-        self._cache = DefaultElementCache()
-        self._graph = SAMMGraph()
-
-    def get_graph(self) -> SAMMGraph:
-        """Get SAMM graph.
+    def __init__(self):
+        self.graph = None
 
-        :return: parsed SAMM Aspect model Graph.
+    def load_aspect_model(self, rdf_graph: Graph, aspect_urn: URIRef | str = "") -> List[Aspect]:
         """
-        return self._graph
-
-    def get_samm_version(self) -> str:
-        """Get SAMM version of the graph."""
-        return self._graph.get_samm_version()
+        Creates a python object(s) to represent the Aspect model graph.
 
-    @staticmethod
-    def convert_file_path(file_path: Union[str, Path]) -> str:
-        """Convert file_path to the string.
-
-        :param file_path: path to model file
-        """
-        if isinstance(file_path, Path):
-            file_path = str(file_path)
+        This function takes an RDF graph and a URN for an Aspect node and converts it into
+        a set of structured and connected Python objects that represents the Aspect model graph. The output is a
+        list of Python objects derived from the RDF graph centered around the specified Aspect node.
 
-        tmp_path = Path(file_path)
-        if not tmp_path.exists():
-            raise FileNotFoundError(f"Could not found the file {tmp_path}")
+        Args:
+            rdf_graph (RDFGraph): The RDF graph from which to create the model.
+            aspect_urn (str): The URN identifier for the main Aspect node in the RDF graph.
 
-        return file_path
+        Returns:
+            list: A list of Python objects that represent the Aspect elements of the Aspect model graph.
 
-    def _reset_graph(self):
-        """Reset graph and cache data."""
-        if self._graph:
-            self._graph = SAMMGraph()
+        Examples:
+            # Assuming 'graph' is a predefined RDFGraph object and 'aspect_urn' is defined:
+            aspect_model = create_aspect_model_graph(graph, "urn:example:aspectNode")
+            print(aspect_model)  # This prints the list of Python objects.
 
-        if self._cache:
-            self._cache = DefaultElementCache()
-
-    def load_aspect_model(self, file_path: Union[Path, str]):
-        """Load aspect model to RDF GRAPH.
-
-        Create an aspect object with all the including properties and operations with the turtle file
-
-        :param file_path: path to the turtle file. Can be either a string or a Path object
-        :return: instance of the aspect
+        Notes:
+            It's crucial that the aspect_urn corresponds to a valid Aspect node within the RDF graph;
+            otherwise, the function may not perform as expected.
         """
-        file_path = self.convert_file_path(file_path)
-        self._reset_graph()
-        _ = self._graph.parse(file_path)
-        loaded_aspect_model = self._graph.to_python()
+        self.graph = SAMMGraph(graph=rdf_graph)
+        loaded_aspect_model = self.graph.to_python(aspect_urn)
+
+        # Add check that loaded_aspect_model is not empty
+        # Add check that aspect_urn is ref to an Aspect node
 
         return loaded_aspect_model

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

@@ -1,4 +1,4 @@
-#  Copyright (c) 2024 Robert Bosch Manufacturing Solutions GmbH
+#  Copyright (c) 2023 Robert Bosch Manufacturing Solutions GmbH
 #
 #  See the AUTHORS file(s) distributed with this work for additional
 #  information regarding authorship.
@@ -9,17 +9,17 @@
 #
 #   SPDX-License-Identifier: MPL-2.0
 
-from pathlib import Path
 from typing import List, Optional, Union
 
 from rdflib import RDF, Graph, URIRef
 from rdflib.graph import Node
 
+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.loader.default_element_cache import DefaultElementCache
 from esmf_aspect_meta_model_python.loader.model_element_factory import ModelElementFactory
-from esmf_aspect_meta_model_python.resolver.base import AspectModelResolver, BaseResolver
+from esmf_aspect_meta_model_python.resolver.meta_model import AspectMetaModelResolver
 from esmf_aspect_meta_model_python.vocabulary.SAMM import SAMM
 
 
@@ -31,16 +31,13 @@ class SAMMGraph:
     def __init__(
         self,
         graph: Graph | None = None,
-        resolver: BaseResolver | None = None,
-        cache: DefaultElementCache | None = None,
+        cache: Union[DefaultElementCache, None] = None,
     ):
-        super().__init__()
-
         self._graph = graph if graph else Graph()
-        self._resolver = resolver if resolver else AspectModelResolver()
         self._cache = cache if cache else DefaultElementCache()
         self._samm_version = ""
-        self._file_path: str = ""
+
+        self.populate_with_meta_data()
 
     def __repr__(self) -> str:
         return repr(self._graph)
@@ -52,7 +49,7 @@ def get_rdf_graph(self) -> Graph:
         """Get RDF graph."""
         return self._graph
 
-    def get_samm_version(self) -> str:
+    def _get_samm_version_from_graph(self):
         """Get SAMM version from the graph."""
         version = ""
 
@@ -63,78 +60,55 @@ def get_samm_version(self) -> str:
 
         return version
 
-    @staticmethod
-    def convert_file_path(file_path: Union[Path, str]) -> str:
-        """Convert file_path to the string.
-
-        :param file_path: path to model file
-        """
-        if isinstance(file_path, Path):
-            file_path = str(file_path)
-
-        return file_path
-
-    def parse(self, file_path: Union[Path, str]) -> Graph:
-        """Parse a file to the SAMM graph.
-
-        :param file_path: Path to the *ttl file.
-        """
-        self._file_path = self.convert_file_path(file_path)
-        self._graph.parse(self._file_path)
-        self._samm_version = self.get_samm_version()
-        self._resolver.resolve(self._graph, self._file_path, self._samm_version)
-
-        return self._graph
+    def get_samm_version(self):
+        """Get SAMM version from the graph."""
+        version = self._get_samm_version_from_graph()
 
-    def _get_model_file_path(self, model_file_path: str = "") -> str:
-        """Get a model file path.
+        if not version:
+            raise ValueError("SAMM version not found in the Graph.")
+        else:
+            self._samm_version = version
 
-        :param model_file_path: str with path to the model
-        :return: validated path rto the model fiel
-        """
-        model_file_path = model_file_path if model_file_path else self._file_path
-        if not model_file_path:
-            raise ValueError("Path to the model is empty")
+    def populate_with_meta_data(self):
+        """Populate RDF graph with SAMM data."""
+        if not self._samm_version:
+            self.get_samm_version()
 
-        return model_file_path
+        meta_model_reader = AspectMetaModelResolver()
+        meta_model_reader.parse(self._graph, self._samm_version)
 
-    def get_nodes_from_graph(self, model_file_path: str = "") -> List[Node]:
-        """Get a list of URIRef to nodes from the base model file."""
+    def get_aspect_nodes_from_graph(self) -> List[Node]:
+        """Get a list of Aspect nodes from the graph."""
         nodes = []
-        model_file_path = self._get_model_file_path(model_file_path)
-        base_graph = Graph().parse(model_file_path, format="turtle")
+        samm = SAMM(self._samm_version)
 
         # Search for Aspect elements
-        samm = SAMM(self._samm_version)
-        for subject in base_graph.subjects(predicate=RDF.type, object=samm.get_urn(SAMM.aspect)):  # type: ignore
+        for subject in self._graph.subjects(predicate=RDF.type, object=samm.get_urn(SAMM.aspect)):  # type: ignore
             nodes.append(subject)
 
-        if not nodes:
-            for subject, object in base_graph.subject_objects(predicate=RDF.type, unique=True):
-                prefix_data = str(object).replace("<", "").split(":")
-                if ":".join(prefix_data[:3]) == self.samm_prefix:
-                    nodes.append(subject)
-
         return nodes
 
     def get_base_nodes(self, aspect_urn: URIRef | str = "") -> List[Node]:
         """Get a list of base graph elements.
 
-        :param aspect_urn: URN of the Aspect node.
+        :param model_pointer: pointer to the model
         :return: List of base graph elements.
         """
         base_elements: list[Node] = []
 
         if aspect_urn:
             base_elements += [aspect_urn if isinstance(aspect_urn, URIRef) else URIRef(aspect_urn)]
         else:
-            base_elements += self.get_nodes_from_graph()
+            base_elements += self.get_aspect_nodes_from_graph()
 
         return base_elements
 
-    def to_python(self, aspect_urn: URIRef | str = "") -> List[URIRef | None]:
+    def to_python(self, aspect_urn: URIRef | str = "") -> List[Aspect]:
         """Convert SAMM graph to Python objects."""
         base_nodes = self.get_base_nodes(aspect_urn)
+        if not base_nodes:
+            raise ValueError(f"Could not found Aspect node by the URN {aspect_urn}.")
+
         model_element_factory = ModelElementFactory(self._samm_version, self._graph, self._cache)
         aspect_elements = model_element_factory.create_all_graph_elements(base_nodes)