Spaces:
Sleeping
Sleeping
File size: 8,564 Bytes
93e1b64 4ae75ec 93e1b64 4ae75ec 1f35211 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 1f35211 646d392 1f35211 646d392 93e1b64 4ae75ec 27d40b9 7833461 27d40b9 7833461 1f35211 4ae75ec 1f35211 4ae75ec 1f35211 4ae75ec 1f35211 4ae75ec 1f35211 4ae75ec 27d40b9 f26b169 27d40b9 f26b169 27d40b9 f26b169 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 4ae75ec 93e1b64 1f35211 4ae75ec 1f35211 4ae75ec |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 |
# %%
from typing import List, Dict, Any
import os
from sqlalchemy import create_engine, text
import requests
from sentence_transformers import SentenceTransformer
username = "demo"
password = "demo"
hostname = os.getenv("IRIS_HOSTNAME", "localhost")
port = "1972"
namespace = "USER"
CONNECTION_STRING = f"iris://{username}:{password}@{hostname}:{port}/{namespace}"
engine = create_engine(CONNECTION_STRING)
def get_all_diseases_name(engine) -> List[List[str]]:
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT * FROM Test.EntityEmbeddings
"""
result = conn.execute(text(sql))
data = result.fetchall()
all_diseases = [row[1] for row in data if row[1] != "nan"]
return all_diseases
def get_uri_from_name(engine, name: str) -> str:
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT uri FROM Test.EntityEmbeddings
WHERE label = '{name}'
"""
result = conn.execute(text(sql))
data = result.fetchall()
return data[0][0].split("/")[-1]
def get_most_similar_diseases_from_uri(
engine, original_disease_uri: str, threshold: float = 0.8
) -> List[str]:
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT * FROM Test.EntityEmbeddings
"""
result = conn.execute(text(sql))
data = result.fetchall()
all_diseases = [row[1] for row in data if row[1] != "nan"]
return all_diseases
def get_uri_from_name(engine, name: str) -> str:
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT uri FROM Test.EntityEmbeddings
WHERE label = '{name}'
"""
result = conn.execute(text(sql))
data = result.fetchall()
return data[0][0].split("/")[-1]
def get_most_similar_diseases_from_uri(
engine, original_disease_uri: str, threshold: float = 0.8
) -> List[str]:
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT TOP 10 e1.uri AS uri1, e2.uri AS uri2, e1.label AS label1, e2.label AS label2,
VECTOR_COSINE(e1.embedding, e2.embedding) AS distance
FROM Test.EntityEmbeddings e1, Test.EntityEmbeddings e2
WHERE e1.uri = 'http://identifiers.org/medgen/{original_disease_uri}'
AND VECTOR_COSINE(e1.embedding, e2.embedding) > {threshold}
AND e1.uri != e2.uri
ORDER BY distance DESC
"""
result = conn.execute(text(sql))
data = result.fetchall()
similar_diseases = [
(row[1].split("/")[-1], row[3], row[4]) for row in data if row[3] != "nan"
]
return similar_diseases
def get_clinical_record_info(clinical_record_id: str) -> Dict[str, Any]:
# Request:
# curl -X GET "https://clinicaltrials.gov/api/v2/studies/NCT00841061" \
# -H "accept: text/csv"
request_url = f"https://clinicaltrials.gov/api/v2/studies/{clinical_record_id}"
response = requests.get(request_url, headers={"accept": "application/json"})
return response.json()
def get_clinical_records_by_ids(clinical_record_ids: List[str]) -> List[Dict[str, Any]]:
clinical_records = []
for clinical_record_id in clinical_record_ids:
clinical_record_info = get_clinical_record_info(clinical_record_id)
clinical_records.append(clinical_record_info)
return clinical_records
def get_similarities_among_diseases_uris(
uri_list: List[str],
) -> List[tuple[str, str, float]]:
uri_list = ", ".join([f"'{uri}'" for uri in uri_list])
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT e1.uri AS uri1, e2.uri AS uri2, VECTOR_COSINE(e1.embedding, e2.embedding) AS distance
FROM Test.EntityEmbeddings e1, Test.EntityEmbeddings e2
WHERE e1.uri IN ({uri_list}) AND e2.uri IN ({uri_list}) AND e1.uri != e2.uri
"""
result = conn.execute(text(sql))
data = result.fetchall()
return data
def augment_the_set_of_diseaces(diseases: List[str]) -> str:
print(diseases)
for i in range(15-len(diseases)):
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT TOP 1 e2.uri AS new_disease, (SUM(VECTOR_COSINE(e1.embedding, e2.embedding))/ {len(diseases)}) AS score
FROM Test.EntityEmbeddings e1, Test.EntityEmbeddings e2
WHERE e1.uri IN ({','.join([f"'{disease}'" for disease in diseases])})
AND e2.uri NOT IN ({','.join([f"'{disease}'" for disease in diseases])})
AND e2.label != 'nan'
GROUP BY e2.label
ORDER BY score DESC
"""
result = conn.execute(text(sql))
data = result.fetchall()
diseases.append(data[0][0].split('/')[-1])
return diseases
def get_embedding(string: str, encoder) -> List[float]:
# Embed the string using sentence-transformers
vector = encoder.encode(string, show_progress_bar=False)
return vector
def get_diseases_related_to_a_textual_description(
description: str, encoder
) -> List[str]:
# Embed the description using sentence-transformers
description_embedding = get_embedding(description, encoder)
string_representation = str(description_embedding.tolist())[1:-1]
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT TOP 5 d.uri, VECTOR_COSINE(d.embedding, TO_VECTOR('{string_representation}', DOUBLE)) AS distance
FROM Test.DiseaseDescriptions d
ORDER BY distance DESC
"""
result = conn.execute(text(sql))
data = result.fetchall()
return [{"uri": row[0], "distance": row[1]} for row in data]
def get_clinical_trials_related_to_diseases(
diseases: List[str], encoder
) -> List[str]:
# Embed the diseases using sentence-transformers
diseases_string = ", ".join(diseases)
disease_embedding = get_embedding(diseases_string, encoder)
string_representation = str(disease_embedding.tolist())[1:-1]
with engine.connect() as conn:
with conn.begin():
sql = f"""
SELECT TOP 5 d.nct_id, VECTOR_COSINE(d.embedding, TO_VECTOR('{string_representation}', DOUBLE)) AS distance
FROM Test.ClinicalTrials d
ORDER BY distance DESC
"""
result = conn.execute(text(sql))
data = result.fetchall()
return [{"nct_id": row[0], "distance": row[1]} for row in data]
if __name__ == "__main__":
username = "demo"
password = "demo"
hostname = os.getenv("IRIS_HOSTNAME", "localhost")
port = "1972"
namespace = "USER"
CONNECTION_STRING = f"iris://{username}:{password}@{hostname}:{port}/{namespace}"
try:
engine = create_engine(CONNECTION_STRING)
diseases = get_most_similar_diseases_from_uri("C1843013")
for disease in diseases:
print(disease)
except Exception as e:
print(e)
try:
print(get_uri_from_name(engine, "Alzheimer disease 3"))
except Exception as e:
print(e)
clinical_record_info = get_clinical_records_by_ids(["NCT00841061"])
print(clinical_record_info)
textual_description = (
"A disease that causes memory loss and other cognitive impairments."
)
encoder = SentenceTransformer("allenai-specter")
diseases = get_diseases_related_to_a_textual_description(
textual_description, encoder
)
for disease in diseases:
print(disease)
try:
similarities = get_similarities_among_diseases_uris(
[
"http://identifiers.org/medgen/C4553765",
"http://identifiers.org/medgen/C4553176",
"http://identifiers.org/medgen/C4024935",
]
)
for similarity in similarities:
print(
f'{similarity[0].split("/")[-1]} and {similarity[1].split("/")[-1]} have a similarity of {similarity[2]}'
)
except Exception as e:
print(e)
# %%
|