-
Notifications
You must be signed in to change notification settings - Fork 0
/
expertawrapper.py
215 lines (195 loc) · 9.99 KB
/
expertawrapper.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
211
212
213
214
215
import numpy as np
from experta import *
class Utils():
def __init__(self):
pass
# we consider suspicion, tumor, nodes, metastasis and cancer stages ready to be determined
# if no relevant requirement is missing (None)
def ready_to_determine(self, my_dict):
for value in my_dict.values():
if value is None:
return False
return True
def determine(self, engine, fact, data):
watch('RULES', 'FACTS')
engine.reset()
engine.declare(fact(**data))
engine.run()
return engine.get_result()
def gaussian(self, min_value, max_value):
mean = (max_value + min_value) / 2.0
std_dev = (max_value - min_value) / 4.0
while True:
sample = np.random.normal(mean, std_dev)
if min_value <= sample <= max_value:
return sample
class Result(Fact):
value = Field(str, default="")
class BaseStage(KnowledgeEngine):
def get_result(self):
for fact in self.facts.values():
if isinstance(fact, Result):
return fact["value"]
return None
# Initialize the Facts
class SUSPICION_FACTS(Fact):
smoking = Field(bool, mandatory=False)
asbestos = Field(bool, mandatory=False)
radio = Field(bool, mandatory=False) # radionucleids
history = Field(bool, mandatory=False)
tsymptoms = Field(bool, mandatory=False)
nsymptoms = Field(bool, mandatory=False)
msymptoms = Field(bool, mandatory=False)
class TUMOR_FACTS(Fact):
mass = Field(bool, mandatory=True) # mass found with x-rays
diameter = Field(float, mandatory=False) # diameter determined with ct-scan
bronchoscopesis = Field(bool, mandatory=False) # bronchoscopesis examination as part of pathologoanatomical examination
cytologic = Field(bool, mandatory=False) # cytologic examination as part of pathologoanatomical examination
nearby_organs = Field(bool, mandatory=False) # infiltracion of nearby organs found with ct-scan
fna_and_pet_scan = Field(bool, mandatory=False) # the results of these two should always agree
class NODES_FACTS(Fact):
lymph_nodes_size = Field(float, mandatory=True)
peribronchial_metastasis = Field(bool, mandatory=False)
mediastinal_metastasis = Field(bool, mandatory=False)
fna_positive = Field(bool, mandatory=False)
class METASTASIS_FACTS(Fact):
separate_tumor_nodules = Field(bool, mandatory=False)
distant_metastasis = Field(bool, mandatory=False)
class TNM_FACTS(Fact):
T = Field(str, mandatory=False)
N = Field(str, mandatory=False)
M = Field(str, mandatory=False)
# Initialize the rules
class SuspicionInvestigation(BaseStage):
@Rule(SUSPICION_FACTS(smoking=False, asbestos=False, radio=False, history=False, tsymptoms=False, nsymptoms=False, msymptoms=False))
def suspicion_does_not_exist(self):
self.declare(Result(value = 'False'))
@Rule(OR(
SUSPICION_FACTS(smoking=True, asbestos=W(), radio=W(), history=W(), tsymptoms=W(), nsymptoms=W(), msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=True, radio=W(), history=W(), tsymptoms=W(), nsymptoms=W(), msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=W(), radio=True, history=W(), tsymptoms=W(), nsymptoms=W(), msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=W(), radio=W(), history=True, tsymptoms=W(), nsymptoms=W(), msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=W(), radio=W(), history=W(), tsymptoms=True, nsymptoms=W(), msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=W(), radio=W(), history=W(), tsymptoms=W(), nsymptoms=True, msymptoms=W()),
SUSPICION_FACTS(smoking=W(), asbestos=W(), radio=W(), history=W(), tsymptoms=W(), nsymptoms=W(), msymptoms=True)))
def suspicion_exists(self):
self.declare(Result(value = 'True'))
class TumorStage(BaseStage):
@Rule(TUMOR_FACTS(mass=False, diameter=P(lambda x: x==0.0),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=False))
def rule_tumor_t0(self):
self.declare(Result(value='T0'))
@Rule(OR(
TUMOR_FACTS(mass=False, diameter=P(lambda x: x==0),
bronchoscopesis=False, cytologic=True, nearby_organs=False, fna_and_pet_scan=False),
TUMOR_FACTS(mass=False, diameter=P(lambda x: x==0),
bronchoscopesis=True, cytologic=False, nearby_organs=False, fna_and_pet_scan=False)))
def rule_tumor_tx(self):
self.declare(Result(value='Tx'))
@Rule(TUMOR_FACTS(mass=True, diameter=P(lambda x: x <= 2),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=True))
def rule_tumor_t1a(self):
self.declare(Result(value='T1a'))
@Rule(TUMOR_FACTS(mass=True, diameter=P(lambda x: 2 < x <= 3),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=True))
def rule_tumor_t1b(self):
self.declare(Result(value='T1b'))
@Rule(TUMOR_FACTS(mass=True, diameter=P(lambda x: 3 < x <= 5),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=True))
def rule_tumor_t2a(self):
self.declare(Result(value='T2a'))
@Rule(TUMOR_FACTS(mass=True, diameter=P(lambda x: 5 < x <= 7),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=True))
def rule_tumor_t2b(self):
self.declare(Result(value='T2b'))
@Rule(TUMOR_FACTS(mass=True, diameter=P(lambda x: x > 7),
bronchoscopesis=W(), cytologic=W(), nearby_organs=False, fna_and_pet_scan=W()))
def rule_tumor_t3(self):
self.declare(Result(value='T3'))
@Rule(TUMOR_FACTS(mass=True, diameter=W(),
bronchoscopesis=W(), cytologic=W(), nearby_organs=True, fna_and_pet_scan=W()))
def rule_tumor_t4(self):
self.declare(Result(value='T4'))
class NodesStage(BaseStage):
@Rule(NODES_FACTS(lymph_nodes_size=P(lambda x: x < 0.5),
peribronchial_metastasis=False, mediastinal_metastasis=False, fna_positive=False))
def rule_nodes_stage_n0(self):
self.declare(Result(value='N0'))
@Rule(NODES_FACTS(lymph_nodes_size=P(lambda x: 0.5 <= x <= 1.5),
peribronchial_metastasis=False, mediastinal_metastasis=False, fna_positive=False))
def rule_nodes_stage_nx(self):
self.declare(Result(value='Nx'))
@Rule(NODES_FACTS(lymph_nodes_size=P(lambda x: x > 1.5),
peribronchial_metastasis=True, mediastinal_metastasis=False, fna_positive=False))
def rule_nodes_stage_n1(self):
self.declare(Result(value='N1'))
@Rule(NODES_FACTS(lymph_nodes_size=P(lambda x: x > 1.5),
peribronchial_metastasis=W(), mediastinal_metastasis=True, fna_positive=False))
def rule_nodes_stage_n2(self):
self.declare(Result(value='N2'))
@Rule(NODES_FACTS(lymph_nodes_size=W(),
peribronchial_metastasis=W(), mediastinal_metastasis=W(), fna_positive=True))
def rule_nodes_stage_n3(self):
self.declare(Result(value='N3'))
class MetastasisStage(BaseStage):
@Rule(METASTASIS_FACTS(separate_tumor_nodules=False, distant_metastasis=False))
def rule_metastasis_stage_m0(self):
self.declare(Result(value='M0'))
@Rule(METASTASIS_FACTS(separate_tumor_nodules=True, distant_metastasis=False))
def rule_metastasis_stage_m1a(self):
self.declare(Result(value='M1a'))
@Rule(METASTASIS_FACTS(separate_tumor_nodules=W(), distant_metastasis=True))
def rule_metastasis_stage_m1b(self):
self.declare(Result(value='M1b'))
class CancerStage(BaseStage):
@Rule(OR(TNM_FACTS(T='Tx', N=W(), M=W()),
TNM_FACTS(T=W(), N='Nx', M=W()),
TNM_FACTS(T=W(), N=W(), M='Mx')))
def rule_stage_x(self):
self.declare(Result(value='Stage Undetermined'))
@Rule(TNM_FACTS(T='T0', N='N0', M='M0'))
def rule_stage_0(self):
self.declare(Result(value='No indication')) # No indication of Lung Cancer was found
@Rule(TNM_FACTS(T='T1a', N='N0', M='M0'))
def rule_stage_ia2(self):
self.declare(Result(value="Stage IA2"))
@Rule(TNM_FACTS(T='T1b', N='N0', M='M0'))
def rule_stage_ia3(self):
self.declare(Result(value="Stage IA3"))
@Rule(OR(TNM_FACTS(T='T1c', N='N0', M='M0'),
TNM_FACTS(T='T2a', N='N0', M='M0')))
def rule_stage_ib(self):
self.declare(Result(value="Stage IB"))
@Rule(OR(TNM_FACTS(T='T2b', N='N0', M='M0'),
TNM_FACTS(T='T3', N='N0', M='M0')))
def rule_stage_iia(self):
self.declare(Result(value="Stage IIA"))
@Rule(OR(TNM_FACTS(T='T1', N='N1', M='M0'),
TNM_FACTS(T='T2', N='N1', M='M0'),
TNM_FACTS(T='T2a', N='N1', M='M0'),
TNM_FACTS(T='T2b', N='N1', M='M0')))
def rule_stage_iib(self):
self.declare(Result(value="Stage IIB"))
@Rule(OR(TNM_FACTS(T='T1', N='N2', M='M0'),
TNM_FACTS(T='T1a', N='N2', M='M0'),
TNM_FACTS(T='T1b', N='N2', M='M0'),
TNM_FACTS(T='T1c', N='N2', M='M0'),
TNM_FACTS(T='T2', N='N2', M='M0'),
TNM_FACTS(T='T2a', N='N2', M='M0'),
TNM_FACTS(T='T2b', N='N2', M='M0'),
TNM_FACTS(T='T3', N='N1', M='M0'),
TNM_FACTS(T='T3', N='N2', M='M0'),
TNM_FACTS(T='T4', N='N1', M='M0'),
TNM_FACTS(T='T4', N='N2', M='M0')))
def rule_stage_iiia(self):
self.declare(Result(value="Stage IIIA"))
@Rule(TNM_FACTS(T=W(), N='N3', M='M0'))
def rule_stage_iiib(self):
self.declare(Result(value="Stage IIIB"))
@Rule(TNM_FACTS(T=W(), N=W(), M='M1a'))
def rule_stage_iiic(self):
self.declare(Result(value="Stage IIIC"))
@Rule(OR(TNM_FACTS(T=W(), N=W(), M='M1b'),
TNM_FACTS(T=W(), N=W(), M='M1c')))
def rule_stage_iv(self):
self.declare(Result(value="Stage IV"))