-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
127 lines (101 loc) · 4.87 KB
/
main.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
'''
script to convert repro scheme to json for fhir questionnaire
Currently supports reproschema to fhir json in en.
#example: "python reprotofhirjson.py <reproschema_folder directory>"
'''
import argparse
from collections import OrderedDict
import json
import os
import shutil
from pathlib import Path
from fhir.resources.questionnaire import Questionnaire
from fhir.resources.valueset import ValueSet
from fhir.resources.codesystem import CodeSystem
from fhir.resources import construct_fhir_element
from reproschema.jsonldutils import load_file
from reproschema_to_fhir.config import Config
from reproschema_to_fhir.fhir import QuestionnaireGenerator
def main():
parser = argparse.ArgumentParser()
# string param of path to folder containing reproschema files
parser.add_argument("reproschema_questionnaire",
type=str,
help="path to folder containing reproschema files")
parser.add_argument("--output",
type=str,
default="output",
help="path to folder to output fhir json")
args = parser.parse_args()
output_path = Path(args.output)
reproschema_folder = Path(args.reproschema_questionnaire)
if not os.path.isdir(reproschema_folder):
raise FileNotFoundError(
f"{reproschema_folder} does not exist. Please check if folder exists and is located at the correct directory"
)
# load each file recursively within the folder into its own key in the reproschema_content dict
reproschema_content = OrderedDict()
for file in reproschema_folder.glob("**/*"):
if file.is_file():
# get the full path to the file *after* the base reproschema_folder path
# since files can be referenced by relative paths, we need to keep track of relative location
filename = str(file.relative_to(reproschema_folder))
with open(f"{reproschema_folder}/{filename}") as f:
reproschema_content[filename] = json.loads(f.read())
schema_name = [
name for name in (reproschema_content.keys())
if name.endswith("_schema")
][0]
reproschema_schema = reproschema_content[schema_name]
if (f"schema:version" in reproschema_schema and
reproschema_schema["schema:version"] not in ("0.0.1", "1.0.0-rc1", "1.0.0")
) or f"schemaVersion" in reproschema_schema and reproschema_schema[
"schemaVersion"] not in ("0.0.1", "1.0.0-rc1", "1.0.0-rc4", "1.0.0"):
raise ValueError(
'Unable to work with reproschema versions other than 0.0.1, 1.0.0-rc1, and 1.0.0-rc4'
)
# before we print to file we wish to validate the jsons using fhir resources
# convert to fhir
config = Config()
questionnaire_generator = QuestionnaireGenerator(config)
fhir_questionnaire = questionnaire_generator.convert_to_fhir(
reproschema_content)
questionnaire_json = construct_fhir_element('Questionnaire',
json.dumps(fhir_questionnaire))
valueset_dict = questionnaire_generator.get_value_set()
for valueset in valueset_dict:
valueset_json = construct_fhir_element('ValueSet',
valueset_dict[valueset])
codesystem_dict = questionnaire_generator.get_code_system()
for codesystem in codesystem_dict:
codesystem_json = construct_fhir_element('CodeSystem',
codesystem_dict[codesystem])
# get filename from the reproschema_folder name provided
file_name = reproschema_folder.parts[-1]
dirpath = Path(output_path / f"{file_name}")
if dirpath.exists() and dirpath.is_dir():
shutil.rmtree(dirpath)
paths = [
output_path / file_name,
output_path / f"{file_name}/valuesets/",
output_path / f"{file_name}/codesystems/"
]
for folder in paths:
folder.mkdir(parents=True, exist_ok=True)
with open(output_path / f"{file_name}/{file_name}.json", "w+") as f:
f.write(json.dumps(fhir_questionnaire))
# write out valuesets and codesystems which have been updated in the generator object
valueset_count = 1
valuesets = [value for (key,value) in questionnaire_generator.get_value_set().items()]
for valueset in valuesets:
with open(output_path / f"{file_name}/valuesets/{file_name}-valueset-{valueset_count}.json", "w+") as f:
f.write(json.dumps(valueset))
valueset_count += 1
codesystem_count = 1
codesystems = [value for (key,value) in questionnaire_generator.get_code_system().items()]
for codesystem in codesystems:
with open(output_path / f"{file_name}/codesystems/{file_name}-codesystem-{codesystem_count}.json", "w+") as f:
f.write(json.dumps(codesystem))
codesystem_count += 1
if __name__ == '__main__':
main()