-
Notifications
You must be signed in to change notification settings - Fork 0
/
get_data.py
210 lines (180 loc) · 6.44 KB
/
get_data.py
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
# 1st Step : Get the data from the UMLS
import os
import sys
import timeit
from umls_api.column_type import ColumnType
from numpy import NaN
import pandas as pd
from umls_api.mysql_connection import db, UMLS_API_KEY
from umls_api.metathesaurus_queries import MetathesaurusQueries
from umls_api.semantic_network_queries import SemanticNetworkQueries
from umls_api.authentication_umls_api import Authentication
def print_description(concept_cui: str, concept_tui: str, concept_gui: str):
"""Print the description of a concept
Args:
concept_cui (str): CUI of the concept
concept_tui (str): TUI of the concept
concept_gui (str): GUI of the concept
"""
print("\n")
print("cui", concept_cui)
print("tui", concept_tui)
print("gui", concept_gui)
print("\n")
def save_index_db(count: int, db_index_path="artefact/db_index"):
"""Save the index of the database in a file
Args:
count (int): Count of the database
db_index_path (str, optional): Path of the file. Defaults to "db_index".
"""
try:
text_file = open(db_index_path, "w+", encoding='utf8')
text_file.write(str(count))
text_file.close()
except Exception as e:
print(str(e))
def save_to_csv(X_data: list, Y_data: list, db_index: int):
"""Save the data to a csv file
Args:
X_data (list): X data
Y_data (list): Y data
db_index (int): Index of the database
"""
data = pd.DataFrame(X_data, columns=['CUI', 'Corpus', 'Has_Definition', 'Prefered_Label',
'Labels', 'SAB', 'Parents_Types_TUI', 'Parents_Types_GUI'])
data = pd.concat(
[data, pd.DataFrame(Y_data, columns=['TUI', 'GUI'])], axis=1)
# Save X and Y in the same csv file
if db_index is not None and db_index > 0:
data.to_csv('artefact/data.csv', mode="a", index=False, header=False)
else:
data.to_csv('artefact/data.csv', index=False)
def save_types():
"""Save all the possible types in two files"""
sty = SemanticNetworkQueries(db)
all_gui = sty.get_all_single_gui()
type_res = [i[0] for i in all_gui]
f = open("artefact/type_gui.txt", "w+")
f.write(" ".join(type_res))
f.close()
all_tui = sty.get_all_tui()
type_res = [i[0] for i in all_tui]
f = open("artefact/type_tui.txt", "w+")
f.write(" ".join(type_res))
f.close()
def generate_source_data(limit: int, config: dict, verbose=False):
"""generate_source_data function"""
db_index = None
db_index_path = "artefact/db_index"
if limit:
if os.path.isfile(db_index_path):
text_file = open(db_index_path, "r", encoding='utf8')
db_index = int(text_file.read())
text_file.close()
print("Index Database: ", db_index)
else:
text_file = open(db_index_path, "w+", encoding='utf8')
text_file.write('0')
text_file.close()
db_index = 0
start = timeit.default_timer()
meta = MetathesaurusQueries(db)
sty = SemanticNetworkQueries(db)
gui_indexes = {}
all_gui = sty.get_all_gui()
for item in all_gui:
gui_indexes[item[1]] = item[0]
res = meta.get_all_unique_terms(nb_data=limit, offset=db_index)
count = 0
X_data = []
Y_data = []
try:
for (cui, tui) in res:
gui = gui_indexes[tui]
if verbose:
print_description(cui, tui, gui)
if verbose:
print("Getting labels...")
try:
prefered_label = meta.get_prefered_label_from_cui(cui)
labels = " ".join(
list(map(lambda x: x[0], meta.get_all_labels_from_cui(cui))))
except Exception as e:
labels = ""
print(str(e))
if verbose:
print("Done.")
if verbose:
print("Getting definition...")
try:
definitions = " ".join(
list(map(lambda x: x[0], meta.get_all_definitions_from_cui(cui))))
has_definition = True
if definitions == "":
has_definition = False
# Corpus is the concatenation of prefered_labels and definitions
corpus = definitions
except Exception as e:
definitions = ""
has_definition = False
print(str(e))
if verbose:
print("Done.")
if verbose:
print("Getting sources...")
try:
sources = meta.get_sources_from_cui(cui)
sab = ""
for i in sources:
if i[0]:
sab += i[0]
if i != sources[-1]:
sab += "/"
except Exception as e:
sab = ""
print(str(e))
if verbose:
print("Done.")
if verbose:
print("Getting parents informations...")
try:
parents_types_TUI = meta.get_all_type_of_parent_from_cui(
cui, ColumnType.TUI , gui_indexes)
parents_types_TUI = " ".join(parents_types_TUI)
parents_types_GUI = meta.get_all_type_of_parent_from_cui(
cui, ColumnType.GUI , gui_indexes)
parents_types_GUI = " ".join(parents_types_GUI)
except Exception as e:
parents_types_TUI = ""
parents_types_GUI = ""
print(str(e))
if verbose:
print("Done.")
row = [
cui,
corpus,
1 if has_definition else 0,
prefered_label,
labels,
sab,
parents_types_TUI,
parents_types_GUI
]
X_data.append(row)
Y_data.append([tui, gui])
count += 1
if verbose:
print("\n")
except KeyboardInterrupt:
print("Quitting...")
save_to_csv(X_data, Y_data, db_index)
if limit:
save_index_db(db_index + count)
sys.exit(0)
save_to_csv(X_data, Y_data, db_index)
if limit:
save_index_db(db_index + count)
save_types()
stop = timeit.default_timer()
print(f'Generate source data in {stop - start} seconds')
return (X_data, Y_data)