1
|
|
|
import csv |
2
|
|
|
from typing import Dict, List |
3
|
|
|
|
4
|
1 |
|
from kerapu import clean_str, clean_int, clean_bool, clean_date |
5
|
|
|
from kerapu.boom.attribuut import maak_attribuut, Attribuut |
6
|
1 |
|
from kerapu.boom.AttribuutGroep import AttribuutGroep |
7
|
1 |
|
from kerapu.boom.attribuut_groep_koppeling import maak_attribuut_groep_koppeling, AttribuutGroepKoppeling |
8
|
1 |
|
from kerapu.boom.BeslisRegel import BeslisRegel |
9
|
1 |
|
from kerapu.boom.ZorgProductGroep import ZorgProductGroep |
10
|
1 |
|
from kerapu.boom.ZorgProductGroepVersie import ZorgProductGroepVersie |
11
|
1 |
|
from kerapu.lbz.Diagnose import Diagnose |
12
|
1 |
|
from kerapu.lbz.Specialisme import Specialisme |
13
|
1 |
|
from kerapu.lbz.Subtraject import Subtraject |
14
|
1 |
|
from kerapu.lbz.ZorgActiviteit import ZorgActiviteit |
15
|
1 |
|
from kerapu.lbz.ZorgType import ZorgType |
16
|
1 |
|
from kerapu.lbz.ZorgVraag import ZorgVraag |
17
|
1 |
|
|
18
|
1 |
|
|
19
|
|
|
class Kerapu: |
20
|
|
|
""" |
21
|
1 |
|
Een implementatie van de grouper in Python. |
22
|
|
|
""" |
23
|
|
|
|
24
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
25
|
|
|
def __init__(self): |
26
|
|
|
""" |
27
|
1 |
|
Object constructor. |
28
|
|
|
""" |
29
|
|
|
self.__zorgproductgroep_boom: Dict[str, ZorgProductGroep] = {} |
30
|
|
|
""" |
31
|
1 |
|
De zorgproductgroepboom. Sleutel is zorgproductgroepcode. |
32
|
|
|
""" |
33
|
|
|
|
34
|
|
|
self.__beslisregels: Dict[int, BeslisRegel] = {} |
35
|
|
|
""" |
36
|
|
|
Alle beslisregels. Sleutel is het ID van de beslisregel. |
37
|
|
|
""" |
38
|
1 |
|
|
39
|
|
|
self.__attribuutgroepen: Dict[int, AttribuutGroep] = {} |
40
|
|
|
""" |
41
|
|
|
Alle attribuutgroepen. Sleutel is het ID van de attribuutgroep. |
42
|
|
|
""" |
43
|
|
|
|
44
|
|
|
self.__attribuut_groep_koppelingen: Dict[int, List[AttribuutGroepKoppeling]] = {} |
45
|
1 |
|
""" |
46
|
|
|
Alle attribuutgroepkoppelingen. Sleutel is het ID van de attribuutgroep. |
47
|
|
|
""" |
48
|
|
|
|
49
|
|
|
self.__attributen: Dict[int, Attribuut] = {} |
50
|
|
|
""" |
51
|
|
|
Alle attributen. Sleutel is het ID van het attribuut. |
52
|
1 |
|
""" |
53
|
|
|
|
54
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
55
|
|
|
def init_static(self, folder: str) -> None: |
56
|
|
|
""" |
57
|
|
|
Initialiseert alle statistische data. |
58
|
|
|
|
59
|
1 |
|
:param str folder: De folder met alle goupertabellen. |
60
|
1 |
|
""" |
61
|
|
|
self.__lees_attribuut_tabel(folder) |
62
|
|
|
self.__lees_attribuut_groep_koppeling_tabel(folder) |
63
|
|
|
self.__lees_attribuut_groepen_tabel(folder) |
64
|
|
|
self.__lees_beslis_regel_tabel(folder) |
65
|
|
|
self.__lees_zorg_product_groepen(folder) |
66
|
|
|
|
67
|
1 |
|
Diagnose.init_static(folder) |
68
|
|
|
Specialisme.init_static(folder) |
69
|
|
|
ZorgActiviteit.init_static(folder) |
70
|
|
|
ZorgType.init_static(folder) |
71
|
|
|
ZorgVraag.init_static(folder) |
72
|
|
|
|
73
|
1 |
|
# ------------------------------------------------------------------------------------------------------------------ |
74
|
1 |
|
def __lees_attribuut_tabel(self, folder: str) -> None: |
75
|
1 |
|
""" |
76
|
1 |
|
Leest de attribuuttabel (opgeslagen in CSV). |
77
|
1 |
|
|
78
|
|
|
:param str folder: De folder met alle groupertabellen in CSV-formaat. |
79
|
1 |
|
""" |
80
|
1 |
|
with open(folder + '/Attributen.csv', 'r', encoding='utf-8') as csv_file: |
81
|
1 |
|
reader = csv.reader(csv_file) |
82
|
1 |
|
regel_nummer = 0 |
83
|
1 |
|
for regel in reader: |
84
|
|
|
regel_nummer += 1 |
85
|
|
|
|
86
|
1 |
|
# Sla de eerste regel met koppen over. |
87
|
|
|
if regel_nummer == 1: |
88
|
|
|
continue |
89
|
|
|
|
90
|
|
|
attribuut_id = int(regel[0]) |
91
|
|
|
boom_parameter_nummer = int(regel[2]) |
92
|
1 |
|
filter_toets_wijze = int(regel[3]) |
93
|
1 |
|
filter_waarde_type = int(regel[4]) |
94
|
1 |
|
onder_filter_waarde = clean_str(regel[5]) |
95
|
1 |
|
boven_filter_waarde = clean_str(regel[6]) |
96
|
1 |
|
|
97
|
|
|
self.__attributen[attribuut_id] = maak_attribuut(attribuut_id, |
98
|
|
|
boom_parameter_nummer, |
99
|
1 |
|
filter_toets_wijze, |
100
|
1 |
|
filter_waarde_type, |
101
|
|
|
onder_filter_waarde, |
102
|
1 |
|
boven_filter_waarde) |
103
|
1 |
|
|
104
|
1 |
|
print("Aantal attributen: %d" % (regel_nummer - 1)) |
105
|
1 |
|
|
106
|
1 |
|
# ------------------------------------------------------------------------------------------------------------------ |
107
|
1 |
|
def __lees_attribuut_groep_koppeling_tabel(self, folder: str) -> None: |
108
|
|
|
""" |
109
|
1 |
|
Leest de attribuutgroepkoppelingen (opgeslagen in CSV). |
110
|
|
|
|
111
|
|
|
:param str folder: De folder met alle groupertabellen in CSV-formaat. |
112
|
|
|
""" |
113
|
|
|
with open(folder + '/AttribuutGroepKoppelingen.csv', 'r', encoding='utf-8') as csv_file: |
114
|
|
|
reader = csv.reader(csv_file) |
115
|
|
|
regel_nummer = 0 |
116
|
1 |
|
for regel in reader: |
117
|
|
|
regel_nummer += 1 |
118
|
|
|
|
119
|
1 |
|
# Sla de eerste regel met koppen over. |
120
|
|
|
if regel_nummer == 1: |
121
|
|
|
continue |
122
|
|
|
|
123
|
|
|
attribuut_groep_id = int(regel[1]) |
124
|
|
|
attribuut_id = int(regel[2]) |
125
|
1 |
|
attribuut_toets_wijze = int(regel[3]) |
126
|
1 |
|
onder_toets_waarde = clean_int(regel[4]) |
127
|
1 |
|
boven_toets_waarde = clean_int(regel[5]) |
128
|
1 |
|
|
129
|
1 |
|
if attribuut_groep_id not in self.__attribuut_groep_koppelingen: |
130
|
|
|
self.__attribuut_groep_koppelingen[attribuut_groep_id] = [] |
131
|
|
|
|
132
|
1 |
|
if attribuut_id not in self.__attributen: |
133
|
1 |
|
raise RuntimeError("Onbekend attribuut: '%d'" % attribuut_id) |
134
|
|
|
|
135
|
1 |
|
koppeling = maak_attribuut_groep_koppeling(attribuut_groep_id, |
136
|
1 |
|
self.__attributen[attribuut_id], |
137
|
1 |
|
attribuut_toets_wijze, |
138
|
1 |
|
onder_toets_waarde, |
139
|
1 |
|
boven_toets_waarde) |
140
|
|
|
|
141
|
1 |
|
self.__attribuut_groep_koppelingen[attribuut_groep_id].append(koppeling) |
142
|
1 |
|
|
143
|
|
|
print("Aantal attribuutgroepkoppelingen: %d" % (regel_nummer - 1)) |
144
|
1 |
|
|
145
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
146
|
|
|
def __lees_attribuut_groepen_tabel(self, folder: str) -> None: |
147
|
1 |
|
""" |
148
|
|
|
Leest de attribuutgroepen (opgeslagen in CSV). |
149
|
|
|
|
150
|
|
|
:param str folder: De folder met alle groupertabellen in CSV-formaat. |
151
|
|
|
""" |
152
|
|
|
with open(folder + '/AttribuutGroepen.csv', 'r', encoding='utf-8') as csv_file: |
153
|
1 |
|
reader = csv.reader(csv_file) |
154
|
|
|
regel_nummer = 0 |
155
|
1 |
|
for regel in reader: |
156
|
|
|
regel_nummer += 1 |
157
|
|
|
|
158
|
1 |
|
# Sla de eerste regel met koppen over. |
159
|
|
|
if regel_nummer == 1: |
160
|
|
|
continue |
161
|
|
|
|
162
|
|
|
attribuut_groep_id = int(regel[0]) |
163
|
|
|
aantal_voorwaarden_voor_true = int(regel[2]) |
164
|
1 |
|
|
165
|
1 |
|
if attribuut_groep_id not in self.__attribuut_groep_koppelingen: |
166
|
1 |
|
raise RuntimeError("Onbekende koppeling: '%d'" % attribuut_groep_id) |
167
|
1 |
|
|
168
|
1 |
|
self.__attribuutgroepen[attribuut_groep_id] = AttribuutGroep(attribuut_groep_id, |
169
|
|
|
aantal_voorwaarden_voor_true, |
170
|
|
|
self.__attribuut_groep_koppelingen[ |
171
|
1 |
|
attribuut_groep_id]) |
172
|
1 |
|
|
173
|
|
|
print("Aantal attributen: %d" % (regel_nummer - 1)) |
174
|
1 |
|
|
175
|
1 |
|
# ------------------------------------------------------------------------------------------------------------------ |
176
|
|
|
def __lees_beslis_regel_tabel(self, folder: str) -> None: |
177
|
1 |
|
""" |
178
|
|
|
Leest de beslisregels (opgeslagen in CSV). |
179
|
|
|
|
180
|
1 |
|
:param str folder: De folder met alle groupertabellen in CSV-formaat. |
181
|
|
|
""" |
182
|
|
|
verrijkingen = {} |
183
|
|
|
with open(folder + '/BeslisRegels.csv', 'r', encoding='utf-8') as csv_file: |
184
|
|
|
reader = csv.reader(csv_file) |
185
|
1 |
|
regel_nummer = 0 |
186
|
|
|
for regel in reader: |
187
|
|
|
regel_nummer += 1 |
188
|
1 |
|
|
189
|
|
|
# Sla de eerste regel met koppen over. |
190
|
|
|
if regel_nummer == 1: |
191
|
|
|
continue |
192
|
|
|
|
193
|
|
|
beslis_regel_id = int(regel[0]) |
194
|
1 |
|
attribuut_groep_id = int(regel[1]) |
195
|
1 |
|
beslist_regel_id_true = clean_int(regel[2]) |
196
|
1 |
|
beslist_regel_id_false = clean_int(regel[3]) |
197
|
1 |
|
label_true = clean_str(regel[4]) |
198
|
1 |
|
label_false = clean_str(regel[5]) |
199
|
1 |
|
indicatie_aanspraakbeperking = clean_bool(regel[6]) |
200
|
|
|
|
201
|
|
|
if attribuut_groep_id not in self.__attribuutgroepen: |
202
|
1 |
|
raise RuntimeError("Onbekende attribuutgroep: '%d'" % attribuut_groep_id) |
203
|
1 |
|
|
204
|
|
|
verrijkingen[beslis_regel_id] = (beslist_regel_id_true, beslist_regel_id_false) |
205
|
1 |
|
|
206
|
1 |
|
self.__beslisregels[beslis_regel_id] = BeslisRegel(beslis_regel_id, |
207
|
1 |
|
self.__attribuutgroepen[attribuut_groep_id], |
208
|
1 |
|
label_true, |
209
|
1 |
|
label_false, |
210
|
1 |
|
indicatie_aanspraakbeperking) |
211
|
1 |
|
|
212
|
|
|
# Verrijk alle beslisregels met beslisregels voor true en false. |
213
|
1 |
|
for beslis_regel_id, (beslist_regel_id_true, beslist_regel_id_false) in verrijkingen.items(): |
214
|
|
|
beslist_regel_true = self.__beslisregels.get(beslist_regel_id_true, None) |
215
|
|
|
beslist_regel_false = self.__beslisregels.get(beslist_regel_id_false, None) |
216
|
1 |
|
self.__beslisregels[beslis_regel_id].verrijk(beslist_regel_true, beslist_regel_false) |
217
|
|
|
|
218
|
1 |
|
print("Aantal beslisregels: %d" % (regel_nummer - 1)) |
219
|
|
|
|
220
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
221
|
|
|
def __lees_zorg_product_groepen(self, folder: str) -> None: |
222
|
|
|
""" |
223
|
|
|
Leest de zorgproductgroepen (opgeslagen in CSV). |
224
|
|
|
|
225
|
1 |
|
:param str folder: De folder met alle groupertabellen in CSV-formaat. |
226
|
1 |
|
""" |
227
|
1 |
|
with open(folder + '/ZorgProductGroepen.csv', 'r', encoding='utf-8') as csv_file: |
228
|
1 |
|
reader = csv.reader(csv_file) |
229
|
|
|
regel_nummer = 0 |
230
|
1 |
|
for regel in reader: |
231
|
|
|
regel_nummer += 1 |
232
|
|
|
|
233
|
1 |
|
# Sla de eerste regel met koppen over. |
234
|
|
|
if regel_nummer == 1: |
235
|
|
|
continue |
236
|
|
|
|
237
|
|
|
zorg_product_groep_code = clean_str(regel[0]) |
238
|
|
|
beslist_regel_id_start = int(regel[2]) |
239
|
1 |
|
begin_datum = clean_date(regel[3]) |
240
|
1 |
|
eind_datum = clean_date(regel[4]) |
241
|
1 |
|
|
242
|
1 |
|
if zorg_product_groep_code not in self.__zorgproductgroep_boom: |
243
|
1 |
|
self.__zorgproductgroep_boom[zorg_product_groep_code] = ZorgProductGroep(zorg_product_groep_code) |
244
|
|
|
|
245
|
|
|
versie = ZorgProductGroepVersie(zorg_product_groep_code, |
246
|
1 |
|
self.__beslisregels[beslist_regel_id_start], |
247
|
1 |
|
begin_datum, |
248
|
|
|
eind_datum) |
249
|
1 |
|
self.__zorgproductgroep_boom[zorg_product_groep_code].versie_toevoegen(versie) |
250
|
1 |
|
|
251
|
1 |
|
print("Aantal zorgproductgroep versies: %d" % (regel_nummer - 1)) |
252
|
1 |
|
|
253
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
254
|
1 |
|
def bepaal_zorg_product_groep(self, subtraject: Subtraject) -> str: |
255
|
1 |
|
""" |
256
|
|
|
Bepaalt de zorgproductgroep van een subtraject. |
257
|
1 |
|
|
258
|
|
|
:param Subtraject subtraject: Het subtraject waarvoor de zorgproductgroep moet worden bepaalt. |
259
|
|
|
|
260
|
|
|
:rtype: str |
261
|
1 |
|
""" |
262
|
|
|
top_boom = self.__zorgproductgroep_boom['0'] |
263
|
1 |
|
subtraject.zorg_product_groep_code = top_boom.klim(subtraject) |
264
|
|
|
|
265
|
|
|
return subtraject.zorg_product_groep_code |
266
|
1 |
|
|
267
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
268
|
|
|
def bepaal_zorg_product(self, subtraject: Subtraject) -> str: |
269
|
|
|
""" |
270
|
|
|
Bepaalt de zorgproduct van een subtraject. |
271
|
|
|
|
272
|
|
|
:param Subtraject subtraject: Het subtraject waarvoor de zorgproductcode moet worden bepaalt. |
273
|
|
|
|
274
|
1 |
|
:rtype: str |
275
|
1 |
|
""" |
276
|
|
|
zorg_product_groep_code = self.bepaal_zorg_product_groep(subtraject) |
277
|
1 |
|
|
278
|
|
|
if zorg_product_groep_code is not None: |
279
|
|
|
top_boom = self.__zorgproductgroep_boom[zorg_product_groep_code] |
280
|
1 |
|
subtraject.zorg_product_code = top_boom.klim(subtraject) |
281
|
|
|
else: |
282
|
|
|
subtraject.zorg_product_code = None |
283
|
|
|
|
284
|
|
|
return subtraject.zorg_product_code |
285
|
|
|
|
286
|
|
|
# ---------------------------------------------------------------------------------------------------------------------- |
287
|
|
|
|