## # This program has been developed by students from the bachelor Computer Science at Utrecht University within the Software Project course. # © Copyright Utrecht University (Department of Information and Computing Sciences) ## import networkx as nx ## # buildGraph builds a NetworkX graph based on the incoming query data # data: Any, the incoming query data # Return: Graph, a NetworkX graph ## def buildGraph(incomingQueryData): graph = nx.Graph() # Add nodes from JSON print(incomingQueryData) for index, attributes in enumerate(incomingQueryData["value"]["payload"]["nodes"]): # Get the node ID from the data like an overly nested dictionary graph.add_nodes_from( [(incomingQueryData["value"]["payload"]["nodes"][index]["id"], attributes)] ) # Add edges from JSON for index, attributes in enumerate(incomingQueryData["value"]["payload"]["edges"]): fr = incomingQueryData["value"]["payload"]["edges"][index]["from"] to = incomingQueryData["value"]["payload"]["edges"][index]["to"] graph.add_edge(fr, to) return graph ## # addNodeMetaData adds machine learning data to a node under a field called mldata for each node # queryData: dict, the incoming query data # metaData: any, the incoming machine learning results # Return: dict, the query result with the added machine learning data ## def addNodeMetaData(incomingQueryData: dict, metaData): for index, attributes in enumerate(incomingQueryData["value"]["payload"]["nodes"]): currId = incomingQueryData["value"]["payload"]["nodes"][index]["id"] if currId in metaData: incomingQueryData["value"]["payload"]["nodes"][index]["mldata"] = metaData[ currId ] return incomingQueryData ## # addNewEdgeMetaData adds machine learning data to a completely new edge under a top-level field as a list of edge objects # queryData: dict, the incoming query data # metaData: any, the incoming machine learning results # Return: dict, the query result with the added machine learning data ## def addNewEdgeMetaData(queryData: dict, metaData): queryData["value"]["mlEdges"] = metaData return queryData ## # optionalCheck checks wether an optional parameter is present in the request # params: dict, the parameters that were in the request propagated from the front-end # Return: dict, the parameters dict modified to be set to None when an optional parameter is left empty ## def optionalCheck(params: dict): for attribute in params: if len(params[attribute]) == 0: params[attribute] = None return params