-
Notifications
You must be signed in to change notification settings - Fork 0
/
example.py
43 lines (37 loc) · 1.27 KB
/
example.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
import sys
from datetime import datetime
from fiber import Cohort # type: ignore
from fiber.condition import ( # type: ignore
Diagnosis,
)
from fiber2xes import (
cohort_to_event_log,
save_event_log_to_file
)
from fiber2xes.filter.condition import Diagnosis as DiagnosisFilter
# CONFIG START
# Fiber filter (See fiber documentation)
cohort_filter = Diagnosis("M54.5", "ICD-10")
cohort_size = 5 # Number of patients in the cohort
trace_type = "mrn" # Trace type should be either "mrn" or "visit"
trace_filter = DiagnosisFilter("M54.5") # Filter for traces in XES log
# CONFIG END
# Query cohort of patients from database
cohort = Cohort(cohort_filter, limit=cohort_size)
print(f'{len(cohort.mrns())} patients in the cohort')
# Create XES event log with fiber2xes
event_log = cohort_to_event_log(
cohort,
trace_type,
trace_filter=trace_filter,
remove_unlisted=True,
verbose=False,
remove_duplicates=True,
abstraction_path="./abstraction-tables/abstraction_high_level.csv",
include_anamnesis_events=False
)
# Write event log to file
dateTimeObj = datetime.now()
timeStr = dateTimeObj.strftime("%Y%m%d-%H%M%S")
save_event_log_to_file(event_log, file_path="./log_" +
timeStr + "_" + trace_type + "_" + str(cohort_size) + ".xes")