-
Notifications
You must be signed in to change notification settings - Fork 1
/
IndexSNOMED.py
178 lines (152 loc) · 6.47 KB
/
IndexSNOMED.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
import os
import time
from os.path import join
from cr.CRIndexKB import CRIndexKB
from index.IndexTerms import IndexTerms
from index.PreprocessSNOMEDTerms import PreprocessSNOMEDTerms
from index.SNOMEDParser import SNOMEDParser
from index.SynonymExpader import SynonymExpader
from util import ConfigConstants
from util.CRConstants import BASE_CLUSTERS, BASE_SYNONYMS, SNOMED_INDEX_FILE
class IndexSNOMED:
resFolder = None
descFile = None
relationsFile = None
outputFolder = None
valid = False
indexConfig = {}
clusters = {}
synClusters = {}
externalSynonyms = {}
rootConcepts = []
crIndexKB = None
def __init__(self, descFile: str, relationsFile: str, outputFolder: str, indexConfig={}):
self.resFolder = 'resources'
self.descFile = descFile
self.relationsFile = relationsFile
self.indexConfig = indexConfig
self.outputFolder = outputFolder
self.valid = False
self.crIndexKB = CRIndexKB()
self.externalSynonyms = {}
self.rootConcepts = []
def index(self):
start = time.time()
self.checkPrerequisites()
if not self.valid:
return
self.loadPrerequisites()
print(' - Preprocessing SNOMED ...')
snomedParser = SNOMEDParser(self.descFile, self.relationsFile, indexConfig=self.indexConfig)
snomedParser.trim(self.rootConcepts)
catDictionary = snomedParser.getCategoriesDictionary()
print(' - Preprocessing SNOMED terms ...')
preprocessOntologyTerms = PreprocessSNOMEDTerms(snomedParser.getSubTree(),
externalSynonyms=self.externalSynonyms,
indexConfig=self.indexConfig)
processedTerms = preprocessOntologyTerms.getProcessedTerms()
print(' - Indexing terms ...')
indexTerms = IndexTerms(processedTerms, self.clusters, self.crIndexKB)
termsToIndex = indexTerms.getTermsToIndex()
voidTokens = indexTerms.getVoidTokens()
synonymExpader = SynonymExpader(termsToIndex, voidTokens, self.synClusters)
termsToIndex = synonymExpader.getTermsToIndex()
print(' - Serializing index ...')
self.crIndexKB.setHPOIndex(termsToIndex)
self.crIndexKB.setCatDictionary(catDictionary)
self.crIndexKB.serialize(join(self.outputFolder, SNOMED_INDEX_FILE), self.clusters)
end = time.time()
print(' - SNOMED index created in {}s'.format(round(end - start, 2)))
def loadPrerequisites(self):
self.loadClusterData()
self.loadSynClusters()
self.loadExternalSynonyms()
def loadClusterData(self):
self.clusters = {}
count = 1
with open(join(self.resFolder, BASE_CLUSTERS), 'r') as f:
lines = f.readlines()
for line in lines:
line = line.strip()
if line:
tokens = line.split(',')
for token in tokens:
self.clusters[token] = 'C' + str(count)
count += 1
def loadSynClusters(self):
self.synClusters = {}
with open(join(self.resFolder, BASE_SYNONYMS), 'r') as f:
lines = f.readlines()
for line in lines:
line = line.strip()
if line:
tokens = line.split(',')
clusterSet = []
for token in tokens:
token = token.strip()
clusterSet.append(self.clusters[token])
for token in tokens:
token = token.strip()
self.synClusters[self.clusters[token]] = clusterSet
def loadExternalSynonyms(self):
if not self.indexConfig:
return
if self.indexConfig:
if not ConfigConstants.VAR_EXTENAL_SYNS in self.indexConfig:
return
with open(self.indexConfig[ConfigConstants.VAR_EXTENAL_SYNS], 'r') as fh:
lines = fh.readlines()
for line in lines:
line = line.strip()
if not line:
continue
segs = line.split('=')
if len(segs) != 2:
continue
uri = segs[0].strip()
if not uri.startswith('SCTID:'):
continue
syn = segs[1].strip()
lst = []
if uri in self.externalSynonyms:
lst = self.externalSynonyms[uri]
lst.append(syn)
self.externalSynonyms[uri] = lst
def checkPrerequisites(self):
if not os.path.isfile(self.descFile):
print('ERROR: SNOMED descriptions file provided [{}] does not exist!'.format(self.descFile))
self.valid = False
return
if not os.path.isfile(self.relationsFile):
print('ERROR: SNOMED relations file provided [{}] does not exist!'.format(self.relationsFile))
self.valid = False
return
if not self.indexConfig:
print('ERROR: SNOMED root concept(s) not provided!')
self.valid = False
return
if not ConfigConstants.VAR_ROOT_CONCEPTS in self.indexConfig:
print('ERROR: SNOMED root concept(s) not provided!')
self.valid = False
return
for rootConcept in self.indexConfig[ConfigConstants.VAR_ROOT_CONCEPTS]:
rootConcept = rootConcept.lower().strip()
rootConcept = rootConcept.replace('sctid', '').strip()
rootConcept = rootConcept.replace(':', '').strip()
if not rootConcept.isnumeric():
print('WARNING: Ignoring SNOMED root concept {} because it is invalid'.format(rootConcept))
continue
self.rootConcepts.append(rootConcept)
if not self.rootConcepts:
print('ERROR: No valid SNOMED root concept(s) found!')
self.valid = False
return
if ConfigConstants.VAR_EXTENAL_SYNS in self.indexConfig:
if not os.path.isfile(self.indexConfig[ConfigConstants.VAR_EXTENAL_SYNS]):
print('WARNING: External synonyms file provided [{}] does not exist!'.format(
self.indexConfig[ConfigConstants.VAR_EXTENAL_SYNS]))
if not os.path.isdir(self.outputFolder):
print('ERROR: Output folder provided [{}] does not exist!'.format(self.outputFolder))
self.valid = False
return
self.valid = True