semanticLinking: added storing nodes for layer

src/data-hub/semantic-linking-microservice/app/configs/swagger.yml

@@ -26,9 +26,34 @@ paths:
           schema:
             type: object
       responses:
-        200:
+        '200':
           description: "Successful echo of request data"
 
+  # nodes region
+  /use-cases/{use_case}/layers/{layer_name}/nodes:
+    get:
+      operationId: "routes.nodes.nodes_for_use_case_and_layer"
+      tags:
+        - "Nodes"
+      summary: "Get all nodes for a Layer"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+        - name: "layer_name"
+          in: "path"
+          description: "Name of the requested Layer"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Node"
+  # endregion nodes
+
   #Raw_dataset region
   /raw_dataset:
     post:
@@ -44,9 +69,9 @@ paths:
             schema:
               $ref: "#/definitions/Dataset"
         responses:
-          201:
+          '201':
             description: "Successful operation"
-          400:
+          '400':
             description: "Invalid input"
     get:
         operationId: "routes.raw-dataset.get"
@@ -55,7 +80,7 @@ paths:
         summary: "Get all datasets"
         parameters: []
         responses:
-          200:
+          '200':
             description: "Successful operation"
             schema:
               $ref: "#/definitions/DatasetCollection"
@@ -73,11 +98,11 @@ paths:
               required: true
               type: "string"
           responses:
-            200:
+            '200':
               description: "Successful operation"
               schema:
                 $ref: "#/definitions/Dataset"
-            404:
+            '404':
               description: "dataset not found"
 
 
@@ -92,11 +117,52 @@ paths:
        summary: "Get all layer data"
        parameters: []
        responses:
-         200:
+         '200':
            description: "Successful operation"
            schema:
              $ref: "#/definitions/LayerCollection"
 
+  /use-cases/{use_case}/layers/{name}:
+    get:
+      operationId: "routes.layers.get_by_name_and_use_case"
+      tags:
+        - "Layers"
+      summary: "Get single layer data"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+        - name: "name"
+          in: "path"
+          description: "Name of the requested layer"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Layer"
+
+  /use-cases/{use_case}/layers:
+    get:
+      operationId: "routes.layers.get_by_use_case"
+      tags:
+        - "Layers"
+      summary: "Get single layer data"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Layer"
+
   /layers/{name}:
      get:
        operationId: "routes.layers.get_by_name"
@@ -110,11 +176,11 @@ paths:
            required: true
            type: "string"
        responses:
-         200:
+         '200':
            description: "Successful operation"
            schema:
              $ref: "#/definitions/Layer"
-         404:
+         '404':
            description: "Layer not found"
 
   /layers/{name}/nodes:
@@ -130,11 +196,11 @@ paths:
            required: true
            type: "string"
        responses:
-         200:
+         '200':
            description: "Successful operation"
            schema:
              $ref: "#/definitions/NodeCollection"
-         404:
+         '404':
            description: "Layer not found"
 
 
@@ -146,7 +212,7 @@ paths:
        summary: "some demo testing"
        parameters: []
        responses:
-         200:
+         '200':
            description: "Successful echo of request data"
 
   /agi/multilayer/multilayer.png:
@@ -159,7 +225,7 @@ paths:
       produces:
         - "image/png"
       responses:
-        200:
+        '200':
           description: "Successful echo of request data"
 
   /graphinfo:
@@ -171,7 +237,7 @@ paths:
          description: "Returns multiple metrics for all nodes created by analyzing and clustering the blockchain traces"
          parameters: []
          responses:
-           200:
+           '200':
              description: "Successful operation"
              schema:
                $ref: "#/definitions/NodeInfo"

src/data-hub/semantic-linking-microservice/app/configs/swagger_local.yml

@@ -29,6 +29,31 @@ paths:
         '200':
           description: "Successful echo of request data"
 
+  # nodes region
+  /use-cases/{use_case}/layers/{layer_name}/nodes:
+    get:
+      operationId: "routes.nodes.nodes_for_use_case_and_layer"
+      tags:
+        - "Nodes"
+      summary: "Get all nodes for a Layer"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+        - name: "layer_name"
+          in: "path"
+          description: "Name of the requested Layer"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Node"
+  # endregion nodes
+
   #Raw_dataset region
   /raw_dataset:
     post:
@@ -97,6 +122,47 @@ paths:
            schema:
              $ref: "#/definitions/LayerCollection"
 
+  /use-cases/{use_case}/layers/{name}:
+    get:
+      operationId: "routes.layers.get_by_name_and_use_case"
+      tags:
+        - "Layers"
+      summary: "Get single layer data"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+        - name: "name"
+          in: "path"
+          description: "Name of the requested layer"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Layer"
+
+  /use-cases/{use_case}/layers:
+    get:
+      operationId: "routes.layers.get_by_use_case"
+      tags:
+        - "Layers"
+      summary: "Get single layer data"
+      parameters:
+        - name: "use_case"
+          in: "path"
+          description: "Name of the requested Use-Case"
+          required: true
+          type: "string"
+      responses:
+        '200':
+          description: "Successful operation"
+          schema:
+            $ref: "#/definitions/Layer"
+
   /layers/{name}:
      get:
        operationId: "routes.layers.get_by_name"

src/data-hub/semantic-linking-microservice/app/db/entities/layer.py

@@ -17,12 +17,25 @@ class Layer:
     def to_serializable_dict(self, for_db=False) -> Dict:
         return {
             "layer_name": self.layer_name,
-            "properties": self.properties
+            "properties": self.properties,
+            "use_case": self.use_case,
+            "total_properties": self.total_properties,
         }
 
+    @staticmethod
+    def from_business_logic_dict(layer_info: Dict):
+        layer = Layer()
+        layer.layer_name = layer_info["name"]
+        layer.properties = layer_info["cluster_properties"]
+        layer.total_properties = layer_info["properties"]
+        layer.use_case = layer_info["use_case"]
+        return layer
+
     def from_serializable_dict(self, layer_info: Dict, from_db=False):
         self.layer_name = layer_info['layer_name']
         self.properties = layer_info['properties']
+        self.use_case = layer_info["use_case"] if "use_case" in layer_info.keys() else None
+        self.total_properties = layer_info["total_properties"] if "total_properties"in layer_info.keys() else None 
 
     def __repr__(self):
         return json.dumps(self.to_serializable_dict())

src/data-hub/semantic-linking-microservice/app/db/repository.py

-import pymongo
 import network_constants as netconst
 from database.MongoRepositoryBase import MongoRepositoryBase
-import json
-
-from db.entities import *
-from typing import List
+from db.entities.layer import Layer
+from db.entities.raw_data import Raw_Dataset
 
+import pymongo
+import json
+from typing import List, Dict
 
 class Repository(MongoRepositoryBase):
     '''This is a repository for MongoDb.'''
@@ -37,12 +37,16 @@ class Repository(MongoRepositoryBase):
 
     # region Layers
     def add_layer(self, layer: Layer):
-        super().insert_entry(self._layer_collection, layer.to_serializable_dict())
+        super().insert_entry(self._layer_collection, layer.to_serializable_dict(for_db=True))
 
     def get_layers(self) -> List[Layer]:
         entries = super().get_entries(self._layer_collection)
         return [Layer(e) for e in entries]
     
+    def get_layers_for_use_case(self, use_case: str) -> List[Layer]:
+        result = super().get_entries(self._layer_collection, projection={'_id': False}, selection={"use_case": use_case})
+        return [Layer(e) for e in result]
+
     def get_layer(self, layer_name) -> Layer:
         entries = super().get_entries(self._layer_collection, selection={'layer_name': layer_name})
         entries = [Layer(e) for e in entries]
@@ -52,6 +56,22 @@ class Repository(MongoRepositoryBase):
         else:
             return None
     
+    def get_layers_by_use_case(self, use_case: str) -> List[Layer]:
+        entries = super().get_entries(self._layer_collection, selection={'use_case': use_case})
+        return [Layer(e) for e in entries]
+
+    def get_layers_by_name_and_use_case(self, layer_name: str, use_case: str) -> List[Layer]:
+        entries = super().get_entries(self._layer_collection, selection={'layer_name': layer_name, 'use_case': use_case})
+        return [Layer(e) for e in entries]
+    
+    def delete_layer(self, layer_name:str):
+        collection = self._database[self._layer_collection]
+        collection.delete_one({"layer_name": layer_name})
+    
+    def delete_layers_for_use_case(self, use_case: str):
+        collection = self._database[self._layer_collection]
+        collection.delete_many({"use_case": use_case})
+
     def add_layer_node(self, node: dict):
         super().insert_entry(self._layer_nodes_collection, node)
 
@@ -64,4 +84,10 @@ class Repository(MongoRepositoryBase):
                                       projection={'_id': 0})
         return [e for e in entries]
     
+    def get_layer_nodes_with_use_case(self, layer_name: str, use_case: str) -> List[Dict]:
+        '''Returns all nodes for the layer.'''
+        entries = super().get_entries(self._layer_nodes_collection, selection={'layer_name': layer_name, 'use_case': use_case},
+                                      projection={'_id': 0})
+        return list(entries)
+
     # endregion

src/data-hub/semantic-linking-microservice/app/messaging/MessageHandler.py

+import network_constants as netconst
+from security.token_manager import TokenManager
+from db.entities import Layer
+from db.repository import Repository
+
 import json
 import requests
-from typing import Dict
+from typing import Dict, List
 from threading import Thread
-import network_constants as netconst
 
 import logging
 LOGGER = logging.getLogger(__name__)
 
 class MessageHandler:
     def __init__(self):
-        pass
+        self._repository = Repository()
 
     def handle_generic(self, body):
         LOGGER.info(f"Received message: {body}")
@@ -32,8 +36,60 @@ class MessageHandler:
         else:
             LOGGER.info("Message Type could not be processed")
     
+    def _fetch_layer_information(self, use_case: str) -> List[Layer]:
+        # fetch token for authentication
+        jwt_token = TokenManager.getInstance().getToken()
+        
+        # query schema information
+        url = f'https://{netconst.BUSINESS_LOGIC_HOSTNAME}:{netconst.BUSINESS_LOGIC_REST_PORT}/api/use-cases/{use_case}/layers'
+
+        response = requests.get(
+                url,
+                verify = False, 
+                proxies = { "http":None, "https":None },
+                headers = { "Authorization": f"Bearer {jwt_token}"}
+        )
+
+        if response.status_code != 200:
+            raise ValueError("no schema information available")
+        
+        layers = [Layer.from_business_logic_dict(row) for row in json.loads(response.text)]
+
+        # update local DB, insert each layer that does not already exists
+        for layer in layers:
+            print(f"Add layer to DB: {layer.to_serializable_dict(for_db=True)}")
+            self._repository.delete_layer(layer.layer_name)
+            self._repository.add_layer(layer)
+
+        return layers
+
     def handle_new_trace(self, content: Dict):
-        pass
+        if "use_case" not in content.keys() or "id" not in content.keys() or "properties" not in content.keys():
+            LOGGER.error(f"Missing fields in trace, required fields: (use_case, id, properties), given fields: ({content.keys()})")
+            return
+        
+        use_case = content["use_case"]
+
+        # fetch layer information
+        layers = self._repository.get_layers_for_use_case(use_case)
+
+        # if no local layers are found, fetch information from server
+        if len(layers) == 0:
+            layers = self._fetch_layer_information(use_case)
+
+        nodes = []
+
+        for layer in layers:
+            node = {}
+                
+            for prop in layer.total_properties:
+                node[prop] = content["properties"][prop]
+
+            node["layer_name"] = layer.layer_name
+            node["use_case"] = use_case
+            nodes.append(node)
+        
+        self._repository.add_layer_nodes(nodes)
 
     def handle_new_traces_available(self):
         # get all traces and call the Processor

src/data-hub/semantic-linking-microservice/app/routes/layers.py

@@ -16,6 +16,12 @@ def get_by_name(name):
     else:
         return Response(status=404)
 
+def get_by_name_and_use_case(name: str, use_case: str):
+    return [layer.to_serializable_dict() for layer in repo.get_layers_by_name_and_use_case(name, use_case)]
+
+def get_by_use_case(use_case: str):
+    return [layer.to_serializable_dict() for layer in repo.get_layers_by_use_case(use_case)]
+
 #endregion
 
 #region nodes

src/data-hub/semantic-linking-microservice/app/routes/nodes.py

+from db.repository import Repository
+from db.entities import Layer
+
+from flask import request, Response
+from typing import List, Dict
+
+repo = Repository()
+
+#region nodes
+
+def nodes_for_use_case_and_layer(layer_name: str, use_case: str) -> List[Dict]:
+    return repo.get_layer_nodes_with_use_case(layer_name, use_case)
\ No newline at end of file

src/data-hub/semantic-linking-microservice/app/tests/test_pipeline.py

+import sys
+import os
+modules_path = '../../../modules/'
+if os.path.exists(modules_path):
+    sys.path.insert(1, modules_path)
+
+import unittest
+import manage_sys_paths
+import json
+
+from messaging.MessageHandler import MessageHandler
+from messaging.DummyMessageManager import DummyMessageManager as DummyMessageSender
+    
+class DummyMongoRepo:
+    '''Dummy class to be used for testing the MessageHandler'''
+    last_trace = None
+    def insert_trace(self, trace):
+        self.last_trace = trace
+
+class Test_HyperGraph(unittest.TestCase):
+    handler = None
+    repo = None
+    msg_sender = None
+
+    def setUp(self):
+        self.repo = DummyMongoRepo()
+        self.msg_sender = DummyMessageSender.get_instance()
+        self.handler = MessageHandler()
+    
+    def _buildTraceMessage(self):
+        return {
+            "type": "new-trace", 
+            "content": {
+                "use_case": "debug", 
+                "id": "b38c916a73747abba1d01dbe11ee15714c90d42b23b8328cafc1667e60045e7b", 
+                "properties": {
+                    "name": "Margherita", 
+                    "dough": False, 
+                    "sauce": "tomato", 
+                    "combined_name": "Margheritawheat", 
+                    "UniqueID": "b38c916a73747abba1d01dbe11ee15714c90d42b23b8328cafc1667e60045e7b", 
+                    "dough_type": "wheat"
+                }
+            }
+        }
+
+    def testTraceProcessing(self):
+        print("START")
+        msg = self._buildTraceMessage()
+        self.handler.handle_new_trace(msg["content"])
+    
+
+if __name__ == '__main__':
+    unittest.main()
\ No newline at end of file