Refactor the model to make it usable without the server

Currently, we can only effectively use the `SemanticMatch` and
`EquivalenceTable` classes with the server. This creates a new
and improved `SemanticMatchDictStore` and refactors the service
to use it.

Furthermore, we clean up the service to use it in a more pythonic
way, eliminating the need for the `service_model` module.

Author: s-heppner

semantic_matcher/model.py

@@ -1,4 +1,5 @@
-from typing import Dict, List
+import json
+from typing import Dict, List, Set, Optional, Iterable
 
 from pydantic import BaseModel
 
@@ -7,15 +8,82 @@ class SemanticMatch(BaseModel):
     """
     A semantic match, mapping two semanticIDs with a matching score. Can be imagined as a weighted graph with
     `base_semantic_id` ---`score`---> `match_semantic_id`
-
-    Todo: Think about static and TTL, but that is optimization
-    Todo: Maybe we want to have the matching method as debug information
     """
     base_semantic_id: str
     match_semantic_id: str
     score: float
     meta_information: Dict
 
+    def __hash__(self):
+        return hash(
+            (self.base_semantic_id, self.match_semantic_id, self.score, frozenset(self.meta_information.items())))
+
+
+class SemanticMatchDictStore:
+    """
+    A collection of `SemanticMatch`es, stored in a Dict, where the Key is the `base_semantic_id` and the Value is
+    the `SemanticMatch` object. This allows for efficient resolution of the `SemanticMatches` of the `base_semantic_id`.
+    """
+    def __init__(self, matches: Iterable[SemanticMatch]):
+        self._store: Dict[str, Set[SemanticMatch]] = {}
+        for x in matches:
+            self.add(x)
+
+    def add(self, match: SemanticMatch) -> None:
+        """
+        Add a `SemanticMatch` to the store
+        """
+        if match.base_semantic_id in self._store:
+            self._store[match.base_semantic_id].add(match)
+        else:
+            self._store[match.base_semantic_id] = {match}
+
+    def discard(self, match: SemanticMatch) -> None:
+        """
+        Discard a `SemanticMatch` from the store
+        """
+        # First we remove the `SemanticMatch` from the set of matches for that `base_semantic_id`
+        self._store[match.base_semantic_id].discard(match)
+        # Then, if there is no more `SemanticMatch`es for that `base_semantic_id`, we remove the Dict entry completely
+        if not len(self._store[match.base_semantic_id]):
+            self._store.pop(match.base_semantic_id)
+
+    def get_all_matches(self) -> Set[SemanticMatch]:
+        """
+        Return a set of all `SemanticMatch`es currently inside the store
+        """
+        all_matches: Set[SemanticMatch] = set()
+        for i in self._store.values():
+            all_matches.update(i)
+        return all_matches
+
+    def get_matches(self, semantic_id: str, min_score: Optional[float] = None) -> Set[SemanticMatch]:
+        """
+        Return all 'SemanticMatches' of a given semantic_id currently inside a store that have a higher or equal
+        score than the `min_score`.
+        """
+        matches = self._store.get(semantic_id, set())
+        return {match for match in matches if min_score is None or match.score >= min_score}
+
+
+    def to_file(self, filename: str) -> None:
+        matches: List[Dict] = [match.model_dump() for match in self.get_all_matches()]
+        with open(filename, "w") as file:
+            json.dump(matches, file, indent=4)
+
+    @classmethod
+    def from_file(cls, filename: str) -> "SemanticMatchDictStore":
+        with open(filename, "r") as file:
+            matches_data = json.load(file)
+        matches = [SemanticMatch(**match_dict) for match_dict in matches_data]
+        return cls(matches)
+
+    def __len__(self) -> int:
+        length = 0
+        for i in self._store.values():
+            length += len(i)
+        return length
+
 
 class EquivalenceTable(BaseModel):
     matches: Dict[str, List[SemanticMatch]]

semantic_matcher/service.py

@@ -1,9 +1,26 @@
-from typing import List
+from typing import Optional, List
 
+from pydantic import BaseModel
 import requests
 from fastapi import APIRouter
 
-from semantic_matcher import model, service_model
+from semantic_matcher import model
+
+
+class MatchRequest(BaseModel):
+    """
+    Request body for the :func:`service.SemanticMatchingService.get_match`
+
+    :ivar semantic_id: The semantic ID that we want to find matches for
+    :ivar local_only: If `True`, only check at the local service and do not request other services
+    :ivar name: Optional name of the resolved semantic ID for NLP matching
+    :ivar definition: Optional definition of the resolved semantic ID for NLP matching
+    """
+    semantic_id: str
+    score_limit: float
+    local_only: bool = True
+    name: Optional[str] = None
+    definition: Optional[str] = None
 
 
 class SemanticMatchingService:
@@ -27,7 +44,7 @@ class SemanticMatchingService:
     def __init__(
             self,
             endpoint: str,
-            equivalences: model.EquivalenceTable
+            matches: model.SemanticMatchDictStore
     ):
         """
         Initializer of :class:`~.SemanticMatchingService`
@@ -46,48 +63,42 @@ def __init__(
         self.router.add_api_route(
             "/get_matches",
             self.get_matches,
+            response_model=List[model.SemanticMatch],
             methods=["GET"]
         )
         self.router.add_api_route(
             "/post_matches",
             self.post_matches,
             methods=["POST"]
         )
-        self.router.add_api_route(
-            "/clear",
-            self.remove_all_matches,
-            methods=["POST"]
-        )
         self.endpoint: str = endpoint
-        self.equivalence_table: model.EquivalenceTable = equivalences
+        self.matches: model.SemanticMatchDictStore = matches
 
     def get_all_matches(self):
         """
         Returns all matches stored in the equivalence table-
         """
-        matches = self.equivalence_table.get_all_matches()
+        matches = self.matches.get_all_matches()
         return matches
 
-    def remove_all_matches(self):
-        self.equivalence_table.remove_all_semantic_matches()
 
     def get_matches(
             self,
-            request_body: service_model.MatchRequest
-    ) -> service_model.MatchesList:
+            request_body: MatchRequest
+    ) -> List[model.SemanticMatch]:
         """
         A query to match two SubmodelElements semantically.
 
         Returns a matching score
         """
         # Try first local matching
-        matches: List[model.SemanticMatch] = self.equivalence_table.get_local_matches(
+        matches: List[model.SemanticMatch] = list(self.matches.get_matches(
             semantic_id=request_body.semantic_id,
-            score_limit=request_body.score_limit
-        )
+            min_score=request_body.score_limit
+        ))
         # If the request asks us to only locally look, we're done already
         if request_body.local_only:
-            return service_model.MatchesList(matches=matches)
+            return matches
         # Now look for remote matches:
         additional_remote_matches: List[model.SemanticMatch] = []
         for match in matches:
@@ -97,7 +108,7 @@ def get_matches(
             remote_matching_service = self._get_matcher_from_semantic_id(match.match_semantic_id)
             if remote_matching_service is None:
                 continue
-            remote_matching_request = service_model.MatchRequest(
+            remote_matching_request = MatchRequest(
                 semantic_id=match.match_semantic_id,
                 # This is a simple "Ungleichung"
                 # Unified score is multiplied: score(A->B) * score(B->C)
@@ -112,20 +123,20 @@ def get_matches(
                 definition=request_body.definition
             )
             url = f"{remote_matching_service}/get_matches"
+            # Todo: Break recursion loop here
             new_matches_response = requests.get(url, json=remote_matching_request.model_dump_json())
-            match_response = service_model.MatchesList.model_validate_json(new_matches_response.text)
-            additional_remote_matches.extend(match_response.matches)
+            response_matches = [model.SemanticMatch(**match) for match in new_matches_response.json()]
+            additional_remote_matches.extend(response_matches)
         # Finally, put all matches together and return
         matches.extend(additional_remote_matches)
-        res = service_model.MatchesList(matches=matches)
-        return res
+        return matches
 
     def post_matches(
             self,
-            request_body: service_model.MatchesList
+            request_body: List[model.SemanticMatch]
     ) -> None:
-        for match in request_body.matches:
-            self.equivalence_table.add_semantic_match(match)
+        for match in request_body:
+            self.matches.add(match)
         # Todo: Figure out how to properly return 200
 
     def _get_matcher_from_semantic_id(self, semantic_id: str) -> str:

semantic_matcher/service_model.py

@@ -0,0 +1,118 @@
+import unittest
+from typing import Dict
+import os
+
+from semantic_matcher.model import SemanticMatch, SemanticMatchDictStore
+
+
+class TestSemanticMatch(unittest.TestCase):
+    def test_creation(self):
+        meta_info: Dict = {"source": "test"}
+        match = SemanticMatch(
+            base_semantic_id="id1",
+            match_semantic_id="id2",
+            score=0.9,
+            meta_information=meta_info
+        )
+        self.assertEqual(match.base_semantic_id, "id1")
+        self.assertEqual(match.match_semantic_id, "id2")
+        self.assertEqual(match.score, 0.9)
+        self.assertEqual(match.meta_information, meta_info)
+
+
+class TestSemanticMatchDictStore(unittest.TestCase):
+    def setUp(self):
+        self.match1 = SemanticMatch(
+            base_semantic_id="id1",
+            match_semantic_id="id2",
+            score=0.9,
+            meta_information={"source": "test1"}
+        )
+        self.match2 = SemanticMatch(
+            base_semantic_id="id1",
+            match_semantic_id="id3",
+            score=0.8,
+            meta_information={"source": "test2"}
+        )
+        self.match3 = SemanticMatch(
+            base_semantic_id="id2",
+            match_semantic_id="id4",
+            score=0.7,
+            meta_information={"source": "test3"}
+        )
+        self.store = SemanticMatchDictStore([self.match1, self.match2, self.match3])
+
+    def test_add(self):
+        match4 = SemanticMatch(
+            base_semantic_id="id3",
+            match_semantic_id="id5",
+            score=0.85,
+            meta_information={"source": "test4"}
+        )
+        self.store.add(match4)
+        self.assertIn(match4, self.store._store["id3"])
+
+    def test_discard(self):
+        self.store.discard(self.match1)
+        self.assertNotIn(self.match1, self.store._store.get("id1", set()))
+
+        self.store.discard(self.match2)
+        self.assertNotIn("id1", self.store._store)  # Since id1 had only match1 and match2, it should be removed
+
+    def test_get_all_matches(self):
+        all_matches = self.store.get_all_matches()
+        self.assertEqual(len(all_matches), 3)
+        self.assertIn(self.match1, all_matches)
+        self.assertIn(self.match2, all_matches)
+        self.assertIn(self.match3, all_matches)
+
+    def test_get_matches(self):
+        # Test without min_score
+        matches = self.store.get_matches("id1")
+        self.assertEqual(len(matches), 2)
+        self.assertIn(self.match1, matches)
+        self.assertIn(self.match2, matches)
+
+        # Test with min_score
+        matches = self.store.get_matches("id1", min_score=0.85)
+        self.assertEqual(len(matches), 1)
+        self.assertIn(self.match1, matches)
+        self.assertNotIn(self.match2, matches)
+
+        # Test with non-existing semantic_id
+        matches = self.store.get_matches("non_existing")
+        self.assertEqual(len(matches), 0)
+
+        # Test with min_score higher than all matches
+        matches = self.store.get_matches("id1", min_score=1.0)
+        self.assertEqual(len(matches), 0)
+
+    def test_to_file_and_from_file(self):
+        filename = "test_store.json"
+
+        self.store.to_file(filename)
+        self.assertTrue(os.path.exists(filename))
+
+        loaded_store = SemanticMatchDictStore.from_file(filename)
+
+        self.assertEqual(len(loaded_store), len(self.store))
+        self.assertEqual(self.store.get_all_matches(), loaded_store.get_all_matches())
+
+        os.remove(filename)  # Clean up test file
+
+    def test_length(self):
+        self.assertEqual(len(self.store), 3)
+        match4 = SemanticMatch(
+            base_semantic_id="id3",
+            match_semantic_id="id5",
+            score=0.85,
+            meta_information={"source": "test4"}
+        )
+        self.store.add(match4)
+        self.assertEqual(len(self.store), 4)
+        self.store.discard(match4)
+        self.assertEqual(len(self.store), 3)
+
+
+if __name__ == "__main__":
+    unittest.main()