Merge pull request #131 from ranking-agent/refactor

Refactor. This cleans out a lot of stuff and makes a functional version of MCQ along with a partial version of EDGAR.

Author: cbizon

documentation/MultiCurieQueries.ipynb

@@ -5,4 +5,4 @@
 # gunicorn --bind 0.0.0.0:6380 -w 1 -k uvicorn.workers.UvicornWorker -t 600 src.server:APP --root-path /1.3
 
 #uvicorn --host 0.0.0.0 --port 6380 --workers 1 src.server:APP --root-path /1.3
-uvicorn --host 0.0.0.0 --port 6380 --workers 1 src.server:APP --root-path /1.4
+uvicorn --host 0.0.0.0 --port 6380 --workers 1 src.server:APP --root-path /

main.sh

@@ -23,7 +23,7 @@ x-translator:
   - Ranking Agent
   infores: "infores:answer-coalesce"
 x-trapi:
-  version: 1.4.0
+  version: 1.5.0
   operations:
   - coalesce answers
   - coalesce by ontology

openapi-config.yaml

@@ -1,7 +1,6 @@
 bmt==1.2.1
-Jinja2~=2.11.2
-jsonschema~=3.2.0
-neo4j~=4.1.1
+Jinja2~=3.1.4
+jsonschema>=4.18.0
 pyyaml~=6.0
 argparse~=1.4.0
 setuptools~=50.3.2
@@ -12,11 +11,12 @@ sparqlwrapper~=1.8.5
 redis~=5.0.3
 fastapi==0.103.1
 sanic~=23.6.0
-httpx~=0.24.1
+httpx>=0.25.0
 gunicorn==21.2.0
 numpy
 pydantic==1.10.10
 uvicorn==0.23.2
 reasoner-pydantic==5.0.2
 jsonlines
 orjson
+pytest-asyncio

requirements.txt

@@ -1,86 +1,218 @@
 from copy import deepcopy
 from collections import defaultdict
-import ast, json
+import ast, json, uuid
 from string import Template
 
-class Opportunity:
-    def __init__(self, hash, qg, kg, a_i, ai2kg):
-        """
-        Define a coalescent opportunity by a hash (the fixed parts of the answers)
-        qg: a (qg_id, semantic type) pair
-        kg: kg_ids allowed for the combined answers
-        a_i: indices into the results saying which answers are combined to give this
-             opportunity
-        kg2a_i: dict mapping frozenset of kg -> which answer they appear in.  Used to filter.
-        """
-        self.answer_hash = hash
-        self.qg_id = qg[0]
-        self.qg_semantic_type = qg[1]
-        self.kg_ids = kg
-        self.answer_indices = a_i
-        self.answerid2kg = ai2kg
-    def get_kg_ids(self):
-        return self.kg_ids
-    def get_qg_id(self):
-        return self.qg_id
-    def get_qg_semantic_type(self):
-        stype = self.qg_semantic_type
-        if isinstance(stype, list):
-            stype = stype[0]
-        return stype
-    def get_answer_indices(self):
-        return self.answer_indices
-    def filter(self,new_kg_ids):
-        """We constructed the opportunities without regard to what nodes we have data on.  Now, we want to filter
-        this opportunity to only answers where we actually have info about the nodes.
-        new_kg_ids is a subset of self.kg_ids.  If we have an answer where we don't have all the nodes, we want to
-        get rid of that answer, and return a new (filtered) opp.  If we get rid of all answers, return None"""
-        if len(new_kg_ids) == len(self.kg_ids):
-            #No filtering required
-            return self
-        nkgids = set(new_kg_ids)
-        new_ai2kg = {}
-        for answer_i, kgs in self.answerid2kg.items():
-            keep = True
-            for kg_i in kgs:
-                if kg_i not in nkgids:
-                    keep = False
-                    break
-            if keep:
-                new_ai2kg[answer_i] = kgs
-        if len(new_ai2kg) == 0:
-            return None
-        #If we removed any answers, we might also need to remove some kg_ids that weren't explicitly filtered
-        final_kg_ids = set()
-        for kgi in new_ai2kg.values():
-            final_kg_ids.update(kgi)
-        return Opportunity(self.answer_hash, (self.qg_id , self.qg_semantic_type ), list(final_kg_ids), list(new_ai2kg.keys()), new_ai2kg)
+###
+# These classes are used to extract the meaning from the TRAPI MCQ query into a more usable form
+###
+
+# TODO: Handle the case where we are not gifted a category for the group node
+class MCQGroupNode:
+    def __init__(self, query_graph):
+        for qnode_id, qnode in query_graph["nodes"].items():
+            if qnode.get("set_interpretation", "") == "MANY":
+                self.curies = qnode["member_ids"]
+                self.qnode_id = qnode_id
+                self.uuid = qnode["ids"][0]
+                self.semantic_type = qnode["categories"][0]
+
+class MCQEnrichedNode:
+    def __init__(self, query_graph):
+        for qnode_id, qnode in query_graph["nodes"].items():
+            if qnode.get("set_interpretation", "") != "MANY":
+                self.qnode_id = qnode_id
+                self.semantic_types = qnode["categories"]
+
+class MCQEdge:
+    def __init__(self, query_graph,groupnode_qnodeid):
+        for qedge_id, qedge in query_graph["edges"].items():
+            if qedge["subject"] == groupnode_qnodeid:
+                self.group_is_subject = True
+            else:
+                self.group_is_subject = False
+            self.qedge_id = qedge_id
+            self.predicate_only = qedge.get("predicates",["biolink:related_to"])[0]
+            self.predicate = {"predicate": self.predicate_only }
+            self.qualifiers = []
+            qualifier_constraints = qedge.get("qualifiers_constraints", [])
+            if len(qualifier_constraints) > 0:
+                qc = qualifier_constraints[0]
+                self.qualifiers = qc.get("qualifier_set", [])
+                for q in self.qualifiers:
+                    self.predicate[q["qualifier_type_id"]] = q["qualifier_value"]
+
+class MCQDefinition:
+    def __init__(self,in_message):
+        query_graph = in_message["message"]["query_graph"]
+        self.group_node = MCQGroupNode(query_graph)
+        self.enriched_node = MCQEnrichedNode(query_graph)
+        self.edge = MCQEdge(query_graph,self.group_node.qnode_id)
+
+
+###
+# These components are about holding the results of Graph enrichment in a TRAPI independent way
+###
 
 class NewNode:
-    def __init__(self,newnode, newnodetype, edge_pred_and_qual, newnode_is, newnode_name):
-        self.newnode = newnode
+    def __init__(self, newnode, newnodetype: list[str]): #edge_pred_and_qual, newnode_is):
+        self.new_curie = newnode
         self.newnode_type = newnodetype
-        self.new_edges = edge_pred_and_qual
-        self.newnode_is = newnode_is
-        self.newnode_name = newnode_name
+        self.newnode_name = None
+
+class NewEdge:
+    def __init__(self, source, predicate: str, target):
+        self.source = source
+        self.predicate = predicate
+        self.target = target
+    def get_prov_link(self):
+        return f"{self.source} {self.predicate} {self.target}"
+
+    def get_sym_prov_link(self):
+        return f"{self.target} {self.predicate} {self.source}"
+
+    def add_prov(self,prov):
+        self.prov = prov
+
+class Lookup_params:
+    def __init__(self, in_message):
+        for qedge_id, qedges in in_message.get("message", {}).get("query_graph", {}).get("edges", {}).items():
+            subject = in_message.get("message", {}).get("query_graph", {}).get("nodes", {})[qedges["subject"]]
+            object = in_message.get("message", {}).get("query_graph", {}).get("nodes", {})[qedges["object"]]
+            if subject.get("ids", []):
+                is_source = True
+            else:
+                is_source = False
+            if is_source:
+                curies = subject["ids"][0]
+                input_qnode = qedges["subject"]
+                output_qnode = qedges["object"]
+                semantic_type = object.get("categories", [])[0]
+            else:
+                curie = object["ids"][0]
+                input_qnode = qedges["object"]
+                output_qnode = qedges["subject"]
+                semantic_type = subject.get("categories", [])[0]
+            predicate_parts = {"predicate": qedges["predicates"][0]}
+            qualifier_constraints = qedges.get("qualifier_constraints", [])
+            if len(qualifier_constraints) > 0:
+                qc = qualifier_constraints[0]
+                qs = qc.get("qualifier_set", [])
+                for q in qs:
+                    predicate_parts[q["qualifier_type_id"].split(":")[1]] = q["qualifier_value"]
+            predicate_parts = json.dumps(predicate_parts, sort_keys=True)
+        self.is_source = is_source
+        self.curie = curie
+        self.predicate_parts = predicate_parts
+        self.input_qnode = input_qnode
+        self.output_qnode = output_qnode
+        self.output_semantic_type = semantic_type
+        self.qedge_id = qedge_id
+
+class Lookup:
+    def __init__(self, curie, predicate, is_source, node_names, node_types, lookup_ids, params_output_semantic_type = []):
+        self.predicate = predicate
+        self.is_source = is_source
+        self.link_ids = lookup_ids
+
+        self.add_input_node(curie, node_types)
+        self.add_input_node_name(node_names)
+        self.add_links(node_names, node_types)
+        self.add_linked_edges(curie, is_source)
+
+
+    def add_input_node(self, curie, node_types):
+        """Optionally, we can patch by adding a new node, which will share a relationship of
+        some sort to the curies in self.set_curies.  The remaining parameters give the edge_type
+        of those edges, as well as defining whether the edge points to the newnode (newnode_is = 'target')
+        or away from it (newnode_is = 'source') """
+        self.input_qnode_curie = NewNode(curie, node_types.get(curie, None))
+    def add_input_node_name(self, node_names):
+        self.input_qnode_curie.name = node_names.get(self.input_qnode_curie.new_curie, None)
+
+    def add_links(self, nodenames, nodetypes):
+        self.lookup_links = [Lookup_Links(link_id, nodenames.get(link_id), nodetypes.get(link_id)) for link_id in self.link_ids]
+    def add_linked_edges(self, input_node, input_node_is_source):
+        """Add edges between the newnode (curie) and the curies that they were linked to"""
+        if input_node_is_source:
+            for i, new_ids in enumerate(self.link_ids):
+                self.lookup_links[i].link_edge = NewEdge(input_node, self.predicate, new_ids)
+        else:
+            for i, new_ids in enumerate(self.link_ids):
+                self.lookup_links[i].link_edge = NewEdge(new_ids, self.predicate, input_node)
+    def add_linked_kg_edges_id(self, eid):
+        """Add edges between the newnode (curie) and the curies that they were linked to as written in the KG"""
+        self.link_kg_edges_ids.append(eid)
+    def get_prov_links(self):
+        return [link.link_edge.get_prov_link() for link in self.lookup_links]
+
+    def add_provenance(self, prov):
+        for link in self.lookup_links:
+            if prov.get(link.link_edge.get_prov_link()):
+                link.link_edge.add_prov(prov[link.link_edge.get_prov_link()])
+            else:
+                link.link_edge.add_prov(prov[link.link_edge.get_sym_prov_link()])
 
-class PropertyPatch:
-    def __init__(self,qg_id,curies,props,answer_ids):
-        self.qg_id = qg_id
-        self.set_curies = curies
-        self.new_props = props
-        self.answer_indices = answer_ids
-        self.added_nodes = []
+    def add_enrichment(self, lookup_indices, enriched_node, predicate, is_source, pvalue):
+        for index in lookup_indices:
+            if hasattr(self.lookup_links[index], 'enrichments'):
+                self.lookup_links[index].enrichments.append(Link_enrichment(enriched_node, predicate, is_source, pvalue))
+            else:
+                self.lookup_links[index].enrichments= [Link_enrichment(enriched_node, predicate, is_source, pvalue)]
+
+class Lookup_Links:
+    def __init__(self, link_id, link_name, link_type):
+        self.link_id = link_id
+        self.link_name = link_name
+        self.link_type = link_type
+
+class Link_enrichment:
+    def __init__(self, enriched_node, predicate, is_source,pvalue):
+        self.enriched_node = enriched_node
+        self.predicate = predicate
+        self.is_source = is_source
+        self.p_value = pvalue
+
+class Enrichment:
+    def __init__(self,p_value,newnode: str, predicate: str, is_source, ndraws, n, total_node_count, curies, node_type: list[str]):
+        """Here the curies are the curies that actually link to newnode, not just the input curies."""
+        self.p_value = p_value
+        self.linked_curies = curies
+        self.enriched_node = None
+        self.predicate = predicate
+        self.is_source = is_source
         self.provmap = {}
-    def add_provenance(self,provmap):
-        self.provmap = provmap
-    def add_extra_node(self,newnode, newnodetype, edge_pred_and_qual, newnode_is,newnode_name):
+        self.add_extra_node(newnode, node_type)
+        self.add_extra_edges(newnode, predicate, is_source)
+        self.counts = [ndraws, n, total_node_count]
+    def add_extra_node(self,newnode, newnodetype: list[str]):
         """Optionally, we can patch by adding a new node, which will share a relationship of
         some sort to the curies in self.set_curies.  The remaining parameters give the edge_type
         of those edges, as well as defining whether the edge points to the newnode (newnode_is = 'target')
         or away from it (newnode_is = 'source') """
-        self.added_nodes.append( NewNode(newnode, newnodetype, edge_pred_and_qual, newnode_is, newnode_name) )
-    def apply(self,answers,question,graph,graph_index,patch_no):
+        self.enriched_node = NewNode(newnode, newnodetype)
+    def add_extra_node_name_and_label(self,name_dict,label_dict):
+        self.enriched_node.newnode_name = name_dict.get(self.enriched_node.new_curie, None)
+        self.enriched_node.newnode_type = label_dict.get(self.enriched_node.new_curie, [])
+    def add_extra_edges(self, newnode, predicate: str, newnode_is_source):
+        """Add edges between the newnode (curie) and the curies that they were linked to"""
+        if newnode_is_source:
+            self.links = [NewEdge(newnode,predicate,curie) for curie in self.linked_curies]
+        else:
+            self.links = [NewEdge(curie,predicate,newnode) for curie in self.linked_curies]
+    def get_prov_links(self):
+        return [link.get_prov_link() for link in self.links]
+    def add_provenance(self,prov):
+        for link in self.links:
+            provlink = link.get_prov_link()
+            symprovlink = link.get_sym_prov_link()
+            if prov.get(provlink):
+                link.add_prov(prov[provlink])
+            else:
+                link.add_prov(prov[symprovlink])
+
+    #TODO: this should not exist in here any more, we are just making a data class
+    def x_apply(self,answers,question,graph,graph_index,patch_no):
         # Find the answers to combine.  It's not necessarily the answer_ids.  Those were the
         # answers that were originally in the opportunity, but we might have only found commonality
         # among a subset of them

src/components.py

@@ -48,10 +48,7 @@ def get_filter_nodes():
     that we don't have anyway, but also a bunch of very generic terms (Disease, Human) that are not useful
     and make lots of links in our database."""
 
-    #At the moment, we are going to get the blocklist from my PR but this can very soon be replaced with
-    # the real one
-    blocklist_url = "https://raw.githubusercontent.com/cbizon/Relay/master/config/blocklist.json"
-    #blocklist_url = "https://raw.githubusercontent.com/NCATSTranslator/Relay/master/config/blocklist.json"
+    blocklist_url = "https://raw.githubusercontent.com/NCATSTranslator/Relay/master/config/blocklist.json"
 
     blocklist = json.loads(requests.get(blocklist_url).text)
     return set(blocklist)