first KG method and edits on py2neo

Author: mandd

src/dackar/knowledge_graph/KGconstruction.py

@@ -16,16 +16,19 @@
 import os, sys
 import tomllib
 import jsonschema
+import copy
 
 class KG:
-    def __init__(self, config_file_path, import_folder_path, uri, pwd, processedDataFolder):
+    #def __init__(self, config_file_path, import_folder_path, uri, pwd, user, processedDataFolder):
+    def __init__(self, config_file_path, uri, pwd, user):    
         # Change import folder to user specific location
-        set_neo4j_import_folder(config_file_path, import_folder_path)
+
+        #set_neo4j_import_folder(config_file_path, import_folder_path)
 
-        self.processedDataFolder = processedDataFolder
+        #self.processedDataFolder = processedDataFolder
 
         # Create python to neo4j driver
-        self.py2neo = Py2Neo(uri=uri, user='neo4j', pwd=pwd)
+        self.py2neo = Py2Neo(uri=uri, user=user, pwd=pwd)
 
         self.graphSchemas = {}
 
@@ -51,23 +54,42 @@ def schemaValidation(self, constructionSchema):
     def genericWorkflow(self, data, constructionSchema):
         # Check constructionSchema against self.graphSchemas  
 
-        # Parse data (pd.dataframe?) and update KG
-
-        '''
-        ---- Example of construction schema ----
-
-            constructionSchema = {'nodes': nodeConstructionSchema,
-                                  'edges': edgeConstructionSchema}
-
-            nodeConstructionSchema = {'nodeLabel1': {'property1': 'node.colA', 'property2': 'node.colB'},
-                                      'nodeLabel2': {'property1': 'node.colC'}}
-            
-            edgeConstructionSchema = [{'source': ('nodeLabel1.property1','col1'),
-                                       'target': ('nodeLabel2.property1','col2'),
-                                       'type': 'edgeType',
-                                       'properties': {'property1': 'colAlpha', 'property2': 'colBeta'}}] 
-        '''
-
+        # Parse data (pd.dataframe) and update KG
+        # Nodes
+        if 'nodes' in constructionSchema:
+            data_temp = copy.deepcopy(data)
+            for node in constructionSchema['nodes'].keys(): 
+                mapping = {value: key for key, value in constructionSchema['nodes'][node].items()}
+                data_renamed = data_temp.rename(columns=mapping)
+                self.py2neo.load_dataframe_for_nodes(df=data_renamed, labels=node, properties=list(mapping.values()))
+        
+        # Relations
+        # --> TODO: check nodes exist
+        if 'relations' in constructionSchema:
+            data_temp = copy.deepcopy(data)
+            for rel in constructionSchema['relations']: 
+                source_node_label = next(iter(rel['source'])).split('.')[0]
+                source_node_prop  = next(iter(rel['source'])).split('.')[1]
+
+                target_node_label = next(iter(rel['target'])).split('.')[0]
+                target_node_prop  = next(iter(rel['target'])).split('.')[1]
+
+                mapping = {}
+                data_renamed = data_temp.rename(columns={next(iter(rel['source'].values())):source_node_prop,
+                                                         next(iter(rel['target'].values())):target_node_prop})
+                
+                for prop in rel['properties'].keys():
+                    data_renamed = data_renamed.rename(columns={rel['properties'][prop]: prop})
+                
+                data_renamed[source_node_label] = source_node_label
+                data_renamed[target_node_label] = target_node_label
+                data_renamed[rel['type']] = rel['type']
+
+                self.py2neo.load_dataframe_for_relations(df=data_renamed, 
+                                                         l1=source_node_label, p1=source_node_prop, 
+                                                         l2=target_node_label, p2=target_node_prop, 
+                                                         lr=rel['type'], 
+                                                         pr=list(rel['properties'].keys()))
 
         '''
         ---- Example of graph schema (toml file) ----
@@ -78,18 +100,32 @@ def genericWorkflow(self, data, constructionSchema):
         # Nodes
         [node.label1]
         description = "This node represents ..."
-        properties = [{name = "prop1", type = "date",   required = bool},
-                      {name = "prop2", type = "string", required = bool},
-        ]
+        properties = {"prop1": {type="date",   required=bool},
+                      "prop2": {type="string", required=bool}}
 
         # Relationships
         [relationships.relation1]
         description = "relation1 indicates ... "
         from_entity = entity1
         to_entity = entity2
-        properties = [{name = "prop1", type = "int"  , required = bool},
-                      {name = "prop2", type = "float", required = bool},
-        ]
+        properties = {"prop1": {type="int",   required=bool},
+                      "prop2": {type="float", required=bool}}
+
+
+
+        ---- Example of construction schema ----
+
+        constructionSchema = {'nodes': nodeConstructionSchema,
+                              'relations': edgeConstructionSchema}
+
+        nodeConstructionSchema = {'nodeLabel1': {'property1': 'node.colA', 'property2': 'node.colB'},
+                                  'nodeLabel2': {'property1': 'node.colC'}}
+        
+        edgeConstructionSchema = [{'source': {'nodeLabel1.property1':'col1'},
+                                   'target': {'nodeLabel2.property1':'col2'},
+                                   'type'  : 'edgeType',
+                                   'properties': {'property1': 'colAlpha', 'property2': 'colBeta'}}] 
+
         '''
 
     # These are workflows specific to the RIAM project
@@ -155,7 +191,7 @@ def monitoringWorkflow(self, filename, constructionSchema):
 
 
     def eventReportWorkflow(self, filename, constructionSchema, pipelines):
-        graphSchemas = TBD
+        graphSchema = TBD
 
         #TODO: Check constructionSchema against graphSchemas
 

src/dackar/knowledge_graph/py2neo.py

@@ -109,18 +109,17 @@ def _create_relation(tx, l1, p1, l2, p2, lr, pr):
         """
         if pr is not None:
             query = f"""
-                MERGE (l1:{l1} {{ {', '.join([f'{k}:"{v}"' for k, v in p1.items()])} }})
-                MERGE (l2:{l2} {{ {', '.join([f'{k}:"{v}"' for k, v in p2.items()])} }})
+                MATCH (l1:{l1} {{ {', '.join([f'{k}:"{v}"' for k, v in p1.items()])} }})
+                MATCH (l2:{l2} {{ {', '.join([f'{k}:"{v}"' for k, v in p2.items()])} }})
                 MERGE (l1)-[r:{lr} {{ {', '.join([f'{k}: ${k}' for k in pr.keys()])} }} ]->(l2)
             """
             tx.run(query, **pr)
         else:
             query = f"""
-                MERGE (l1:{l1} {{ {', '.join([f'{k}:"{v}"' for k, v in p1.items()])} }})
-                MERGE (l2:{l2} {{ {', '.join([f'{k}:"{v}"' for k, v in p2.items()])} }})
+                MATCH (l1:{l1} {{ {', '.join([f'{k}:"{v}"' for k, v in p1.items()])} }})
+                MATCH (l2:{l2} {{ {', '.join([f'{k}:"{v}"' for k, v in p2.items()])} }})
                 MERGE (l1)-[r:{lr}]->(l2)
             """
-            # print(query)
             tx.run(query)
 
     def find_nodes(self, label, properties=None):
@@ -278,7 +277,7 @@ def load_dataframe_for_nodes(self, df, labels, properties):
         """
         assert set(properties).issubset(set(df.columns))
         for _, row in df.iterrows():
-            self.create_node(labels, row.to_dict())
+            self.create_node(labels, row[properties].to_dict())
 
     # Load csv function to create relations
     def load_dataframe_for_relations(self, df, l1='sourceLabel', p1='sourceNodeId', l2='targetLabel', p2='targetNodeId', lr='relationshipType', pr=None):
@@ -297,17 +296,19 @@ def load_dataframe_for_relations(self, df, l1='sourceLabel', p1='sourceNodeId',
         # for nodes. Future development need to be performed.
         # label (l1/l2), properties (p1/p2), and relation label (lr), relation properties (pr)
         valid = []
+
         valid.extend([l1, l2, lr, p1, p2])
         if pr is not None:
             valid.extend(pr)
+
         assert set(valid).issubset(set(df.columns))
 
         with self.__driver.session() as session:
             for _, row in df.iterrows():
                 l1_ = row[l1]
-                p1_ = {'nodeId': row[p1]}
+                p1_ = {p1: row[p1]}
                 l2_ = row[l2]
-                p2_ = {'nodeId': row[p2]}
+                p2_ = {p2: row[p2]}
                 lr_ = row[lr]
                 if pr is not None:
                     pr_ = row[pr].to_dict()

src/dackar/knowledge_graph/schemas/monitoringSystemSchema.toml

@@ -6,11 +6,11 @@ version = "1.0"
 # ====================
 
 [entities.monitoring_variable]
-description = "Represents a measurable physical variable that is continuously or periodically recorded by monitoring sensors from 
-               a specific component, equipment, 
-               or system. This entity captures dynamic operational data that reflects the real-time or historical 
-               state of the asset, enabling monitoring, diagnostics, performance analysis, and integration with 
-               predictive models or digital twins within engineering and plant operations contexts."
+description = "Represents a measurable physical variable that is continuously or periodically recorded by monitoring 
+               sensors from a specific component, equipment, or system. This entity captures dynamic operational 
+               data that reflects the real-time or historical state of the asset, enabling monitoring, diagnostics, 
+               performance analysis, and integration with predictive models or digital twins within engineering and 
+               plant operations contexts."
 properties = [
     { name = "label", type = "string", optional = true },
     { name = "ID",    type = "string"},