Merge pull request #36 from bci-oss/add_get_units_function

Added SAMM meta model class for units

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

@@ -65,3 +65,30 @@ This script can be executed with
 poetry run download-samm-branch
 ```
 to download and start working with the Aspect Model Loader.
+
+## Semantic Aspect Meta Model
+
+To be able to use SAMM meta model classes you need to download the corresponding files. 
+Details are described in [Set up SAMM Aspect Meta Model for development](#set-up-samm-aspect-meta-model-for-development).
+
+This module contains classes for working with Aspect data.
+
+SAMM meta model contains:
+- SammUnitsGraph: provide a functionality for working with units([units.ttl](./esmf_aspect_meta_model_python/samm_aspect_meta_model/samm/unit/2.1.0/units.ttl)) data
+
+### SammUnitsGraph
+
+The class contains functions for accessing units of measurement.
+```python 
+from esmf_aspect_meta_model_python.samm_meta_model import SammUnitsGraph
+
+units_graph = SammUnitsGraph()
+unit_data = units_graph.get_info("unit:volt")
+# {'preferredName': rdflib.term.Literal('volt', lang='en'), 'commonCode': rdflib.term.Literal('VLT'), ... }
+
+units_graph.print_description(unit_data)
+# preferredName: volt
+# commonCode: VLT
+# ...
+# symbol: V
+```

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

@@ -0,0 +1,97 @@
+#  Copyright (c) 2024 Robert Bosch Manufacturing Solutions GmbH
+#
+#  See the AUTHORS file(s) distributed with this work for additional
+#  information regarding authorship.
+#
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+#
+#   SPDX-License-Identifier: MPL-2.0
+
+from os.path import exists, join
+from pathlib import Path
+from string import Template
+from typing import Dict, Union
+
+import rdflib
+
+
+class SammUnitsGraph:
+    """Model units graph."""
+
+    SAMM_VERSION = "2.1.0"
+    UNIT_FILE_PATH = f"samm_aspect_meta_model/samm/unit/{SAMM_VERSION}/units.ttl"
+    QUERY_TEMPLATE = Template("SELECT ?key ?value WHERE {$unit ?key ?value .}")
+
+    def __init__(self):
+        self.unit_file_path = self._get_file_path()
+        self._validate_path()
+        self._graph = self._get_units()
+
+    @property
+    def graph(self) -> rdflib.Graph:
+        """Getter for the units graph."""
+        return self._graph
+
+    def _get_file_path(self) -> str:
+        """Get a path to the units.ttl file"""
+        base_path = Path(__file__).resolve()
+        file_path = join(base_path.parents[0], self.UNIT_FILE_PATH)
+
+        return file_path
+
+    def _validate_path(self):
+        """Checking the path to the units.ttl file."""
+        if not exists(self.unit_file_path):
+            raise ValueError(f"There is no such file {self.unit_file_path}")
+
+    def _get_units(self) -> rdflib.Graph:
+        """Parse a units to graph."""
+        graph = rdflib.Graph()
+        graph.parse(self.unit_file_path, format="turtle")
+
+        return graph
+
+    def _get_nested_data(self, value: str) -> tuple[str, Union[str, Dict]]:
+        """Get data of the nested node."""
+        node_type = value.split("#")[1]
+        node_value: Union[str, Dict] = value
+
+        if node_type != "Unit":
+            node_value = self.get_info(f"unit:{node_type}")
+
+        return node_type, node_value
+
+    def get_info(self, unit: str) -> Dict:
+        """Get a description of the unit."""
+        unit_data: Dict = {}
+        query = self.QUERY_TEMPLATE.substitute(unit=unit)
+        res = self._graph.query(query)
+
+        for row in res:
+            key = row.key.split("#")[1]
+            value = row.value
+            if isinstance(value, rdflib.term.URIRef):
+                sub_key, value = self._get_nested_data(value)
+                if key != "type":
+                    unit_data.setdefault(key, []).append({sub_key: value})
+            else:
+                unit_data[key] = value
+
+        return unit_data
+
+    def print_description(self, unit_data: Dict, tabs: int = 0):
+        """Pretty print a unit data."""
+        for key, value in unit_data.items():
+            if isinstance(value, dict):
+                print("\t" * tabs + f"{key}:")
+                self.print_description(value, tabs + 1)
+            elif isinstance(value, list):
+                print("\t" * tabs + f"{key}:")
+                for node in value:
+                    for key, sub_value in node.items():
+                        print("\t" * (tabs + 1) + f"{key}:")
+                        self.print_description(sub_value, tabs + 2)
+            else:
+                print("\t" * tabs + f"{key}: {value}")

core/esmf-aspect-meta-model-python/tests/unit/test_samm_meta_model.py

@@ -0,0 +1,145 @@
+"""SAMM Meta Model functions test suite."""
+
+from unittest import mock
+
+import pytest
+
+from rdflib.term import URIRef
+
+from esmf_aspect_meta_model_python.samm_meta_model import SammUnitsGraph
+
+
+class TestSammCli:
+    """SAMM Units Graph tests."""
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_init(self, get_file_path_mock, validate_path_mock, get_units_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        get_units_mock.return_value = "graph"
+        result = SammUnitsGraph()
+
+        assert result.graph == "graph"
+        get_file_path_mock.assert_called_once()
+        validate_path_mock.assert_called_once()
+        get_units_mock.assert_called_once()
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.join")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.Path")
+    def test_get_file_path(self, path_mock, join_mock, _, get_units_mock):
+        base_path_mock = mock.MagicMock()
+        base_path_mock.parents = ["parent", "child"]
+        path_mock.return_value = path_mock
+        path_mock.resolve.return_value = base_path_mock
+        join_mock.return_value = "file_path"
+        get_units_mock.return_value = "graph"
+        result = SammUnitsGraph()
+
+        assert result.unit_file_path == "file_path"
+        path_mock.assert_called_once()
+        path_mock.resolve.assert_called_once()
+        join_mock.assert_called_once_with("parent", SammUnitsGraph.UNIT_FILE_PATH)
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.exists")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_validate_path(self, get_file_path_mock, exists_mock, get_units_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        exists_mock.return_value = True
+        get_units_mock.return_value = "graph"
+        result = SammUnitsGraph()
+
+        assert result is not None
+        exists_mock.assert_called_once_with("unit_file_path")
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.exists")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_validate_path_with_exception(self, get_file_path_mock, exists_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        exists_mock.return_value = False
+        with pytest.raises(ValueError) as error:
+            SammUnitsGraph()
+
+        assert str(error.value) == "There is no such file unit_file_path"
+        exists_mock.assert_called_once_with("unit_file_path")
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.rdflib.Graph")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_get_units(self, get_file_path_mock, validate_path_mock, rdflib_graph_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        graph_mock = mock.MagicMock()
+        rdflib_graph_mock.return_value = graph_mock
+        result = SammUnitsGraph()
+
+        assert result._graph == graph_mock
+        rdflib_graph_mock.assert_called_once()
+        graph_mock.parse.assert_called_once_with("unit_file_path", format="turtle")
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_get_nested_data_unit(self, get_file_path_mock, _, get_units_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        get_units_mock.return_value = "graph"
+        units_graph = SammUnitsGraph()
+        result = units_graph._get_nested_data("prefix#Unit")
+
+        assert result == ("Unit", "prefix#Unit")
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph.get_info")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_get_nested_data_not_unit(self, get_file_path_mock, _, get_units_mock, get_info_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        get_units_mock.return_value = "graph"
+        get_info_mock.return_value = "nested_value"
+        units_graph = SammUnitsGraph()
+        result = units_graph._get_nested_data("prefix#unitType")
+
+        assert result == ("unitType", "nested_value")
+        get_info_mock.assert_called_once_with("unit:unitType")
+
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_nested_data")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.isinstance")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_units")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._validate_path")
+    @mock.patch("esmf_aspect_meta_model_python.samm_meta_model.SammUnitsGraph._get_file_path")
+    def test_get_info(self, get_file_path_mock, _, get_units_mock, isinstance_mock, get_nested_data_mock):
+        get_file_path_mock.return_value = "unit_file_path"
+        isinstance_mock.side_effect = (False, URIRef, URIRef)
+        get_nested_data_mock.side_effect = [("type_key", "type_description"), ("sub_unit", "sub_unit_description")]
+        row_1_mock = mock.MagicMock()
+        row_1_mock.key = "prefix#unitType"
+        row_1_mock.value = "unit_1"
+        row_2_mock = mock.MagicMock()
+        row_2_mock.key = "prefix#type"
+        row_2_mock.value = "unit_2"
+        row_3_mock = mock.MagicMock()
+        row_3_mock.key = "prefix#otherUnit"
+        row_3_mock.value = "unit_3"
+        graph_mock = mock.MagicMock()
+        graph_mock.query.return_value = [row_1_mock, row_2_mock, row_3_mock]
+        get_units_mock.return_value = graph_mock
+        units_graph = SammUnitsGraph()
+        result = units_graph.get_info("unit:unit_name")
+
+        assert "unitType" in result
+        assert result["unitType"] == "unit_1"
+        assert "otherUnit" in result
+        assert len(result["otherUnit"]) == 1
+        assert "sub_unit" in result["otherUnit"][0]
+        assert result["otherUnit"][0]["sub_unit"] == "sub_unit_description"
+        graph_mock.query.assert_called_once_with("SELECT ?key ?value WHERE {unit:unit_name ?key ?value .}")
+        isinstance_mock.assert_has_calls(
+            [
+                mock.call("unit_1", URIRef),
+                mock.call("unit_2", URIRef),
+                mock.call("unit_3", URIRef),
+            ]
+        )
+        get_nested_data_mock.assert_has_calls([mock.call("unit_2"), mock.call("unit_3")])